Allowed changes to the SSH config while gapis is running.

This will periodically poll for SSH devices, and update the
file that contains that data as necessary.

cmd/gapis/main.go

@@ -124,12 +124,8 @@ func run(ctx context.Context) error {
 	}
 
 	if *remoteSSHConfig != "" {
-		f, err := os.Open(*remoteSSHConfig)
-		if err != nil {
-			return err
-		}
 		wg.Add(1)
-		crash.Go(func() { getRemoteSSHDevices(ctx, r, f, wg.Done) })
+		crash.Go(func() { monitorRemoteSSHDevices(ctx, r, wg.Done) })
 	}
 
 	deviceScanDone, onDeviceScanDone := task.NewSignal()
@@ -173,15 +169,36 @@ func monitorAndroidDevices(ctx context.Context, r *bind.Registry, scanDone func(
 	}
 }
 
-func getRemoteSSHDevices(ctx context.Context, r *bind.Registry, f io.Reader, scanDone func()) {
-	// Populate the registry with all the existing devices.
-	defer scanDone() // Signal that we have a primed registry.
+func monitorRemoteSSHDevices(ctx context.Context, r *bind.Registry, scanDone func()) {
+	getRemoteSSHConfig := func() (io.ReadCloser, error) {
+		f, err := os.Open(*remoteSSHConfig)
+		if err != nil {
+			return nil, err
+		}
+		return f, nil
+	}
 
-	if devs, err := remotessh.Devices(ctx, f); err == nil {
-		for _, d := range devs {
-			r.AddDevice(ctx, d)
-			r.SetDeviceProperty(ctx, d, client.LaunchArgsKey, text.SplitArgs(*gapirArgStr))
+	func() {
+		// Populate the registry with all the existing devices.
+		defer scanDone() // Signal that we have a primed registry.
+
+		f, err := getRemoteSSHConfig()
+		if err != nil {
+			log.E(ctx, "Could not open remote ssh config")
+			return
 		}
+		defer f.Close()
+
+		if devs, err := remotessh.Devices(ctx, f); err == nil {
+			for _, d := range devs {
+				r.AddDevice(ctx, d)
+				r.SetDeviceProperty(ctx, d, client.LaunchArgsKey, text.SplitArgs(*gapirArgStr))
+			}
+		}
+	}()
+
+	if err := remotessh.Monitor(ctx, r, time.Second*15, getRemoteSSHConfig); err != nil {
+		log.W(ctx, "Could not scan for remote SSH devices. Error: %v", err)
 	}
 }
 

core/os/android/adb/device.go

@@ -217,7 +217,7 @@ func scanDevices(ctx context.Context) error {
 
 	for serial, status := range parsed {
 		cached, ok := cache[serial]
-		if !ok || status != cached.Status() {
+		if !ok || status != cached.Status(ctx) {
 			device, err := newDevice(ctx, serial, status)
 			if err != nil {
 				return err
@@ -299,6 +299,10 @@ func (b *binding) NativeBridgeABI(ctx context.Context, emulated *device.ABI) *de
 	return native
 }
 
+func (b *binding) IsLocal(ctx context.Context) (bool, error) {
+	return true, nil
+}
+
 var abiToISAs = []struct {
 	abi *device.ABI
 	isa string

core/os/device/bind/device.go

@@ -26,7 +26,7 @@ type Device interface {
 	// Instance returns the instance information for this device.
 	Instance() *device.Instance
 	// State returns the last known connected status of the device.
-	Status() Status
+	Status(ctx context.Context) Status
 	// Shell is a helper that builds a shell.Cmd with d.ShellTarget() as its target
 	Shell(name string, args ...string) shell.Cmd
 	// TempFile creates a temporary file on the given Device. It returns the
@@ -53,6 +53,8 @@ type Device interface {
 	IsDirectory(ctx context.Context, path string) (bool, error)
 	// GetWorkingDirectory returns the directory that this device considers CWD
 	GetWorkingDirectory(ctx context.Context) (string, error)
+	// IsLocal returns true if this tracer is local
+	IsLocal(ctx context.Context) (bool, error)
 	// CanTrace returns true if this device can be used to take a trace
 	CanTrace() bool
 }

core/os/device/bind/simple.go

@@ -51,7 +51,7 @@ func (b *Simple) CanTrace() bool { return true }
 func (b *Simple) Instance() *device.Instance { return b.To }
 
 // Status implements the Device interface returning the Status from the LastStatus field.
-func (b *Simple) Status() Status { return b.LastStatus }
+func (b *Simple) Status(ctx context.Context) Status { return b.LastStatus }
 
 // Shell implements the Device interface returning commands that will error if run.
 func (b *Simple) Shell(name string, args ...string) shell.Cmd {
@@ -135,6 +135,10 @@ func (b *Simple) GetWorkingDirectory(ctx context.Context) (string, error) {
 	return os.Getwd()
 }
 
+func (b *Simple) IsLocal(ctx context.Context) (bool, error) {
+	return true, nil
+}
+
 // ABI implements the Device interface returning the first ABI from the Information, or UnknownABI if it has none.
 func (b *Simple) ABI() *device.ABI {
 	if len(b.To.Configuration.ABIs) <= 0 {

core/os/device/remotessh/commands.go

@@ -30,6 +30,7 @@ import (
 	"github.com/google/gapid/core/event/task"
 	"github.com/google/gapid/core/log"
 	"github.com/google/gapid/core/os/device"
+	"github.com/google/gapid/core/os/device/bind"
 	"github.com/google/gapid/core/os/shell"
 	"github.com/google/gapid/core/text"
 )
@@ -132,6 +133,14 @@ func (t sshShellTarget) String() string {
 	return c.User + "@" + c.Host + ": " + t.b.String()
 }
 
+func (b binding) Status(ctx context.Context) bind.Status {
+	_, err := b.Shell("echo", "Hello World").Call(ctx)
+	if err != nil {
+		return bind.Status_Offline
+	}
+	return bind.Status_Online
+}
+
 // Shell implements the Device interface returning commands that will error if run.
 func (b binding) Shell(name string, args ...string) shell.Cmd {
 	return shell.Command(name, args...).On(sshShellTarget{&b})

core/os/device/remotessh/device.go

@@ -23,9 +23,12 @@ import (
 	"net"
 	"os"
 	"strings"
+	"sync"
+	"time"
 
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/google/gapid/core/app/layout"
+	"github.com/google/gapid/core/event/task"
 	"github.com/google/gapid/core/log"
 	"github.com/google/gapid/core/os/device"
 	"github.com/google/gapid/core/os/device/bind"
@@ -51,6 +54,11 @@ type Device interface {
 	DefaultReplayCacheDir() string
 }
 
+const (
+	// Frequency at which to print scan errors
+	printScanErrorsEveryNSeconds = 120
+)
+
 // MaxNumberOfSSHConnections defines the max number of ssh connections to each
 // ssh remote device that can be used to run commands concurrently.
 const MaxNumberOfSSHConnections = 15
@@ -124,24 +132,112 @@ func (b *binding) newPooledSession() (*pooledSession, error) {
 	}, nil
 }
 
+// Interface check
 var _ Device = &binding{}
 
-// Devices returns the list of reachable SSH devices.
+var (
+	// Registry of all the discovered devices.
+	registry = bind.NewRegistry()
+
+	// cache is a map of device names to fully resolved bindings.
+	cache      = map[string]*binding{}
+	cacheMutex sync.Mutex // Guards cache.
+)
+
+// Monitor updates the registry with devices that are added and removed at the
+// specified interval. Monitor returns once the context is cancelled.
+func Monitor(ctx context.Context, r *bind.Registry, interval time.Duration, conf func() (io.ReadCloser, error)) error {
+	unlisten := registry.Listen(bind.NewDeviceListener(r.AddDevice, r.RemoveDevice))
+	defer unlisten()
+
+	for _, d := range registry.Devices() {
+		r.AddDevice(ctx, d)
+	}
+
+	var lastErrorPrinted time.Time
+	for {
+		rc, err := conf()
+		if err != nil {
+			return err
+		}
+
+		configurations, err := ReadConfigurations(rc)
+		rc.Close()
+		if err != nil {
+			return err
+		}
+
+		if err := scanDevices(ctx, configurations); err != nil {
+			if time.Since(lastErrorPrinted).Seconds() > printScanErrorsEveryNSeconds {
+				log.E(ctx, "Couldn't scan devices: %v", err)
+				lastErrorPrinted = time.Now()
+			}
+		} else {
+			lastErrorPrinted = time.Time{}
+		}
+
+		select {
+		case <-task.ShouldStop(ctx):
+			return nil
+		case <-time.After(interval):
+		}
+	}
+}
+
+// Devices returns the list of attached Android devices.
 func Devices(ctx context.Context, configuration io.Reader) ([]bind.Device, error) {
 	configurations, err := ReadConfigurations(configuration)
 	if err != nil {
+		log.E(ctx, "%+v\n\n", err)
+		return nil, err
+	}
+	log.E(ctx, "%+v\n\n", configurations)
+
+	if err := scanDevices(ctx, configurations); err != nil {
 		return nil, err
 	}
+	devs := registry.Devices()
+	out := make([]bind.Device, len(devs))
+	for i, d := range devs {
+		out[i] = d
+	}
+	return out, nil
+}
 
-	devices := make([]bind.Device, 0, len(configurations))
+func scanDevices(ctx context.Context, configurations []Configuration) error {
+	cacheMutex.Lock()
+	defer cacheMutex.Unlock()
+	allConfigs := make(map[string]bool)
 
 	for _, cfg := range configurations {
-		if device, err := GetConnectedDevice(ctx, cfg); err == nil {
-			devices = append(devices, device)
+		allConfigs[cfg.Name] = true
+
+		// If this device already exists, see if we
+		// can/have to remove it
+		if cached, ok := cache[cfg.Name]; ok {
+			if !deviceStillConnected(ctx, cached) {
+				delete(cache, cfg.Name)
+				registry.RemoveDevice(ctx, cached)
+			}
+		} else {
+			if device, err := GetConnectedDevice(ctx, cfg); err == nil {
+				registry.AddDevice(ctx, device)
+				cache[cfg.Name] = device.(*binding)
+			}
 		}
 	}
 
-	return devices, nil
+	for name, dev := range cache {
+		if _, ok := allConfigs[name]; !ok {
+			delete(cache, name)
+			registry.RemoveDevice(ctx, dev)
+		}
+	}
+	return nil
+}
+
+func deviceStillConnected(ctx context.Context, d *binding) bool {
+	return d.Status(ctx) == bind.Status_Online
 }
 
 // getSSHAgent returns a connection to a local SSH agent, if one exists.

gapis/trace/desktop/trace.go

@@ -56,9 +56,14 @@ func (t *DesktopTracer) TraceConfiguration(ctx context.Context) (*service.Device
 		return nil, err
 	}
 
+	isLocal, err := t.b.IsLocal(ctx)
+	if err != nil {
+		return nil, err
+	}
+
 	return &service.DeviceTraceConfiguration{
 		Apis:                 apis,
-		ServerLocalPath:      true,
+		ServerLocalPath:      isLocal,
 		CanSpecifyCwd:        true,
 		CanUploadApplication: false,
 		CanSpecifyEnv:        true,

test/robot/replay/client.go

@@ -66,9 +66,9 @@ func (c *client) onDeviceAdded(ctx context.Context, host *device.Instance, targe
 			target.Shell("am", "force-stop", "com.google.android.gapid.armeabi").Run(ctx)
 		}()
 		defer job.UnlockDevice(ctx, target)
-		if target.Status() != bind.Status_Online {
+		if target.Status(ctx) != bind.Status_Online {
 			log.I(ctx, "Trying to replay %s on %s not started, device status %s",
-				t.Input.Trace, target.Instance().GetSerial(), target.Status().String())
+				t.Input.Trace, target.Instance().GetSerial(), target.Status(ctx).String())
 			return nil
 		}
 		return c.replay(ctx, t)

test/robot/report/client.go

@@ -66,9 +66,9 @@ func (c *client) onDeviceAdded(ctx context.Context, host *device.Instance, targe
 			target.Shell("am", "force-stop", "com.google.android.gapid.armeabi").Run(ctx)
 		}()
 		defer job.UnlockDevice(ctx, target)
-		if target.Status() != bind.Status_Online {
+		if target.Status(ctx) != bind.Status_Online {
 			log.I(ctx, "Trying to report %s on %s not started, device status %s",
-				t.Input.Trace, target.Instance().GetSerial(), target.Status().String())
+				t.Input.Trace, target.Instance().GetSerial(), target.Status(ctx).String())
 			return nil
 		}
 		return c.report(ctx, t)

test/robot/trace/client.go

@@ -65,9 +65,9 @@ func (c *client) onDeviceAdded(ctx context.Context, host *device.Instance, targe
 	traceOnTarget := func(ctx context.Context, t *Task) error {
 		job.LockDevice(ctx, target)
 		defer job.UnlockDevice(ctx, target)
-		if target.Status() != bind.Status_Online {
+		if target.Status(ctx) != bind.Status_Online {
 			log.I(ctx, "Trying to trace %s on %s not started, device status %s",
-				t.Input.Subject, target.Instance().GetSerial(), target.Status().String())
+				t.Input.Subject, target.Instance().GetSerial(), target.Status(ctx).String())
 			return nil
 		}
 		return c.trace(ctx, target, t)