From 34f33cf4122e6dd6f670574ae3b90e0e7c0dea10 Mon Sep 17 00:00:00 2001 From: Silvano Cerza Date: Wed, 16 Feb 2022 17:24:21 +0100 Subject: [PATCH 1/2] Discoveries are now closed and unregistered after failure --- arduino/discovery/discovery.go | 17 ++++------------- arduino/discovery/discovery_client/main.go | 4 +--- .../discoverymanager/discoverymanager.go | 17 ++++++++++++++--- commands/board/list.go | 11 +++++------ 4 files changed, 24 insertions(+), 25 deletions(-) diff --git a/arduino/discovery/discovery.go b/arduino/discovery/discovery.go index 4af87a0bddf..1ce6310f156 100644 --- a/arduino/discovery/discovery.go +++ b/arduino/discovery/discovery.go @@ -374,21 +374,12 @@ func (disc *PluggableDiscovery) Stop() error { } // Quit terminates the discovery. No more commands can be accepted by the discovery. -func (disc *PluggableDiscovery) Quit() error { - if err := disc.sendCommand("QUIT\n"); err != nil { - return err - } - if msg, err := disc.waitMessage(time.Second * 10); err != nil { - return fmt.Errorf(tr("calling %[1]s: %[2]w"), "QUIT", err) - } else if msg.EventType != "quit" { - return errors.Errorf(tr("communication out of sync, expected '%[1]s', received '%[2]s'"), "quit", msg.EventType) - } else if msg.Error { - return errors.Errorf(tr("command failed: %s"), msg.Message) - } else if strings.ToUpper(msg.Message) != "OK" { - return errors.Errorf(tr("communication out of sync, expected '%[1]s', received '%[2]s'"), "OK", msg.Message) +func (disc *PluggableDiscovery) Quit() { + _ = disc.sendCommand("QUIT\n") + if _, err := disc.waitMessage(time.Second * 5); err != nil { + logrus.Errorf("Quitting discovery %s: %s", disc.id, err) } disc.killProcess() - return nil } // List executes an enumeration of the ports and returns a list of the available diff --git a/arduino/discovery/discovery_client/main.go b/arduino/discovery/discovery_client/main.go index 41651df3592..8a78765f63c 100644 --- a/arduino/discovery/discovery_client/main.go +++ b/arduino/discovery/discovery_client/main.go @@ -135,9 +135,7 @@ out: } for _, disc := range discoveries { - if err := disc.Quit(); err != nil { - log.Fatal("Error stopping discovery:", err) - } + disc.Quit() fmt.Println("Discovery QUITed") for disc.State() == discovery.Alive { time.Sleep(time.Millisecond) diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go index 97a881ad37e..fc89b8ff13a 100644 --- a/arduino/discovery/discoverymanager/discoverymanager.go +++ b/arduino/discovery/discoverymanager/discoverymanager.go @@ -22,6 +22,7 @@ import ( "github.com/arduino/arduino-cli/arduino/discovery" "github.com/arduino/arduino-cli/i18n" "github.com/pkg/errors" + "github.com/sirupsen/logrus" ) // DiscoveryManager is required to handle multiple pluggable-discovery that @@ -64,6 +65,14 @@ func (dm *DiscoveryManager) Add(disc *discovery.PluggableDiscovery) error { return nil } +// remove quits and deletes the discovery with specified id +// from the discoveries managed by this DiscoveryManager +func (dm *DiscoveryManager) remove(id string) { + dm.discoveries[id].Quit() + delete(dm.discoveries, id) + logrus.Infof("Closed and removed discovery %s", id) +} + // parallelize runs function f concurrently for each discovery. // Returns a list of errors returned by each call of f. func (dm *DiscoveryManager) parallelize(f func(d *discovery.PluggableDiscovery) error) []error { @@ -103,6 +112,7 @@ func (dm *DiscoveryManager) RunAll() []error { } if err := d.Run(); err != nil { + dm.remove(d.GetID()) return fmt.Errorf(tr("discovery %[1]s process not started: %[2]w"), d.GetID(), err) } return nil @@ -119,6 +129,7 @@ func (dm *DiscoveryManager) StartAll() []error { return nil } if err := d.Start(); err != nil { + dm.remove(d.GetID()) return fmt.Errorf(tr("starting discovery %[1]s: %[2]w"), d.GetID(), err) } return nil @@ -139,6 +150,7 @@ func (dm *DiscoveryManager) StartSyncAll() (<-chan *discovery.Event, []error) { eventCh, err := d.StartSync(5) if err != nil { + dm.remove(d.GetID()) return fmt.Errorf(tr("start syncing discovery %[1]s: %[2]w"), d.GetID(), err) } @@ -170,6 +182,7 @@ func (dm *DiscoveryManager) StopAll() []error { } if err := d.Stop(); err != nil { + dm.remove(d.GetID()) return fmt.Errorf(tr("stopping discovery %[1]s: %[2]w"), d.GetID(), err) } return nil @@ -185,9 +198,7 @@ func (dm *DiscoveryManager) QuitAll() []error { return nil } - if err := d.Quit(); err != nil { - return fmt.Errorf(tr("quitting discovery %[1]s: %[2]w"), d.GetID(), err) - } + d.Quit() return nil }) return errs diff --git a/commands/board/list.go b/commands/board/list.go index fcfec9296b2..3c5524a377a 100644 --- a/commands/board/list.go +++ b/commands/board/list.go @@ -296,15 +296,14 @@ func Watch(instanceID int32, interrupt <-chan bool) (<-chan *rpc.BoardListWatchR Error: boardsError, } case <-interrupt: - errs := dm.StopAll() - if len(errs) > 0 { + for _, err := range dm.StopAll() { + // Discoveries that return errors have their process + // closed and are removed from the list of discoveries + // in the manager outChan <- &rpc.BoardListWatchResponse{ EventType: "error", - Error: tr("stopping discoveries: %s", errs), + Error: tr("stopping discoveries: %s", err), } - // Don't close the channel if quitting all discoveries - // failed, otherwise some processes might be left running. - continue } return } From 8064d108cfa8f56e837cb3bc5fcec87943fdd776 Mon Sep 17 00:00:00 2001 From: Silvano Cerza Date: Thu, 17 Feb 2022 14:50:18 +0100 Subject: [PATCH 2/2] Add mutex to guard discoveries in DiscoveryManager --- .../discoverymanager/discoverymanager.go | 32 +++++++++++++++++-- 1 file changed, 30 insertions(+), 2 deletions(-) diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go index fc89b8ff13a..b18fbf46bbd 100644 --- a/arduino/discovery/discoverymanager/discoverymanager.go +++ b/arduino/discovery/discoverymanager/discoverymanager.go @@ -28,7 +28,8 @@ import ( // DiscoveryManager is required to handle multiple pluggable-discovery that // may be shared across platforms type DiscoveryManager struct { - discoveries map[string]*discovery.PluggableDiscovery + discoveriesMutex sync.Mutex + discoveries map[string]*discovery.PluggableDiscovery } var tr = i18n.Tr @@ -43,12 +44,16 @@ func New() *DiscoveryManager { // Clear resets the DiscoveryManager to its initial state func (dm *DiscoveryManager) Clear() { dm.QuitAll() + dm.discoveriesMutex.Lock() + defer dm.discoveriesMutex.Unlock() dm.discoveries = map[string]*discovery.PluggableDiscovery{} } // IDs returns the list of discoveries' ids in this DiscoveryManager func (dm *DiscoveryManager) IDs() []string { ids := []string{} + dm.discoveriesMutex.Lock() + defer dm.discoveriesMutex.Unlock() for id := range dm.discoveries { ids = append(ids, id) } @@ -58,6 +63,8 @@ func (dm *DiscoveryManager) IDs() []string { // Add adds a discovery to the list of managed discoveries func (dm *DiscoveryManager) Add(disc *discovery.PluggableDiscovery) error { id := disc.GetID() + dm.discoveriesMutex.Lock() + defer dm.discoveriesMutex.Unlock() if _, has := dm.discoveries[id]; has { return errors.Errorf(tr("pluggable discovery already added: %s"), id) } @@ -68,8 +75,11 @@ func (dm *DiscoveryManager) Add(disc *discovery.PluggableDiscovery) error { // remove quits and deletes the discovery with specified id // from the discoveries managed by this DiscoveryManager func (dm *DiscoveryManager) remove(id string) { - dm.discoveries[id].Quit() + dm.discoveriesMutex.Lock() + d := dm.discoveries[id] delete(dm.discoveries, id) + dm.discoveriesMutex.Unlock() + d.Quit() logrus.Infof("Closed and removed discovery %s", id) } @@ -78,7 +88,13 @@ func (dm *DiscoveryManager) remove(id string) { func (dm *DiscoveryManager) parallelize(f func(d *discovery.PluggableDiscovery) error) []error { var wg sync.WaitGroup errChan := make(chan error) + dm.discoveriesMutex.Lock() + discoveries := []*discovery.PluggableDiscovery{} for _, d := range dm.discoveries { + discoveries = append(discoveries, d) + } + dm.discoveriesMutex.Unlock() + for _, d := range discoveries { wg.Add(1) go func(d *discovery.PluggableDiscovery) { defer wg.Done() @@ -215,7 +231,13 @@ func (dm *DiscoveryManager) List() ([]*discovery.Port, []error) { Port *discovery.Port } msgChan := make(chan listMsg) + dm.discoveriesMutex.Lock() + discoveries := []*discovery.PluggableDiscovery{} for _, d := range dm.discoveries { + discoveries = append(discoveries, d) + } + dm.discoveriesMutex.Unlock() + for _, d := range discoveries { wg.Add(1) go func(d *discovery.PluggableDiscovery) { defer wg.Done() @@ -254,7 +276,13 @@ func (dm *DiscoveryManager) List() ([]*discovery.Port, []error) { // ListCachedPorts return the current list of ports detected from all discoveries func (dm *DiscoveryManager) ListCachedPorts() []*discovery.Port { res := []*discovery.Port{} + dm.discoveriesMutex.Lock() + discoveries := []*discovery.PluggableDiscovery{} for _, d := range dm.discoveries { + discoveries = append(discoveries, d) + } + dm.discoveriesMutex.Unlock() + for _, d := range discoveries { if d.State() != discovery.Syncing { // Discovery is not syncing continue