veyron/services/mgmt/node/impl: consolidate the callback logic in one place,
exposing a narrower API to the rest of the node manager so invokers can listen
for values from children without being exposed to the internals.
The API is in the form of a listener interface, that hides the details of
callback IDs, channels, etc.
Tangentially related to the above, I added logic to the unit test to check that
callback channels get cleaned up properly. Did this using a generic 'Leaking'
method that's only compiled for tests and does not expose the implementation of
the rest of the node manager (we can add more verification items to Leaking
later on). Also changed the tests to stop the node manager before the conclusion
of each test (Cleanup kills the child rather than shutting it down cleanly). To
achieve clean shutdown, I went the route of exposing the child's PID via stdout
and then sending it a catchable signal.
Change-Id: I39b782261807306965ea65bed9b438eaea0de484
diff --git a/services/mgmt/node/impl/app_invoker.go b/services/mgmt/node/impl/app_invoker.go
index 1f6a616..db89956 100644
--- a/services/mgmt/node/impl/app_invoker.go
+++ b/services/mgmt/node/impl/app_invoker.go
@@ -347,52 +347,44 @@
}
// Setup up the child process callback.
callbackState := i.callback
- id := callbackState.generateID()
+ listener := callbackState.listenFor(mgmt.AppCycleManagerConfigKey)
+ defer listener.cleanup()
cfg := config.New()
- cfg.Set(mgmt.ParentNodeManagerConfigKey, naming.MakeTerminal(naming.Join(i.config.Name, configSuffix, id)))
+ cfg.Set(mgmt.ParentNodeManagerConfigKey, listener.name())
handle := vexec.NewParentHandle(cmd, vexec.ConfigOpt{cfg})
- // Make the channel buffered to avoid blocking the Set method when
- // nothing is receiving on the channel. This happens e.g. when
- // unregisterCallbacks executes before Set is called.
- callbackChan := make(chan string, 1)
- callbackState.register(id, mgmt.AppCycleManagerConfigKey, callbackChan)
- defer callbackState.unregister(id)
// Start the child process.
if err := handle.Start(); err != nil {
vlog.Errorf("Start() failed: %v", err)
return nil, errOperationFailed
}
// Wait for the child process to start.
- testTimeout := 10 * time.Second
- if err := handle.WaitForReady(testTimeout); err != nil {
- vlog.Errorf("WaitForReady(%v) failed: %v", testTimeout, err)
+ timeout := 10 * time.Second
+ if err := handle.WaitForReady(timeout); err != nil {
+ vlog.Errorf("WaitForReady(%v) failed: %v", timeout, err)
if err := handle.Clean(); err != nil {
vlog.Errorf("Clean() failed: %v", err)
}
return nil, errOperationFailed
}
- select {
- case childName := <-callbackChan:
- instanceInfo := &instanceInfo{
- AppCycleMgrName: childName,
- Pid: handle.Pid(),
- }
- if err := saveInstanceInfo(instanceDir, instanceInfo); err != nil {
- if err := handle.Clean(); err != nil {
- vlog.Errorf("Clean() failed: %v", err)
- }
- return nil, err
- }
- // TODO(caprita): Spin up a goroutine to reap child status upon
- // exit and transition it to suspended state if it exits on its
- // own.
- case <-time.After(testTimeout):
- vlog.Errorf("Waiting for callback timed out")
+ childName, err := listener.waitForValue(timeout)
+ if err != nil {
if err := handle.Clean(); err != nil {
vlog.Errorf("Clean() failed: %v", err)
}
return nil, errOperationFailed
}
+ instanceInfo := &instanceInfo{
+ AppCycleMgrName: childName,
+ Pid: handle.Pid(),
+ }
+ if err := saveInstanceInfo(instanceDir, instanceInfo); err != nil {
+ if err := handle.Clean(); err != nil {
+ vlog.Errorf("Clean() failed: %v", err)
+ }
+ return nil, err
+ }
+ // TODO(caprita): Spin up a goroutine to reap child status upon exit and
+ // transition it to suspended state if it exits on its own.
return []string{instanceID}, nil
}
