From 1a5dfc5f860f3ddde463f660e52566a585488bdc Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Wed, 13 Jul 2022 14:53:38 +0200
Subject: [PATCH 01/15] Improved streaming of pluggable-discoveries events
 (WIP)

Now the DiscoveryManager is able to start the discoveries and add/remove
them in a thread-safe way. Also the watchers may connect and disconnect
seamlessly at any time, the incoming events from the discovery are
broadcasted correctly to each active watcher.

This refactoring dramatically simplifies the DiscoveryManager design.
---
 .../discoverymanager/discoverymanager.go      | 297 +++++++-----------
 cli/arguments/port.go                         |  25 +-
 cli/board/list.go                             |   3 +-
 commands/board/list.go                        | 109 ++-----
 commands/daemon/daemon.go                     |  27 +-
 5 files changed, 149 insertions(+), 312 deletions(-)

diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index b18fbf46bbd..e6857e77de0 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -28,8 +28,12 @@ import (
 // DiscoveryManager is required to handle multiple pluggable-discovery that
 // may be shared across platforms
 type DiscoveryManager struct {
-	discoveriesMutex sync.Mutex
-	discoveries      map[string]*discovery.PluggableDiscovery
+	discoveriesMutex   sync.Mutex
+	discoveries        map[string]*discovery.PluggableDiscovery
+	discoveriesRunning bool
+	feed               chan *discovery.Event
+	watchersMutex      sync.Mutex
+	watchers           map[*PortWatcher]bool
 }
 
 var tr = i18n.Tr
@@ -38,15 +42,20 @@ var tr = i18n.Tr
 func New() *DiscoveryManager {
 	return &DiscoveryManager{
 		discoveries: map[string]*discovery.PluggableDiscovery{},
+		watchers:    map[*PortWatcher]bool{},
+		feed:        make(chan *discovery.Event, 50),
 	}
 }
 
 // Clear resets the DiscoveryManager to its initial state
 func (dm *DiscoveryManager) Clear() {
-	dm.QuitAll()
 	dm.discoveriesMutex.Lock()
-	defer dm.discoveriesMutex.Unlock()
+	for _, d := range dm.discoveries {
+		d.Quit()
+		logrus.Infof("Closed and removed discovery %s", d.GetID())
+	}
 	dm.discoveries = map[string]*discovery.PluggableDiscovery{}
+	dm.discoveriesMutex.Unlock()
 }
 
 // IDs returns the list of discoveries' ids in this DiscoveryManager
@@ -60,233 +69,137 @@ func (dm *DiscoveryManager) IDs() []string {
 	return ids
 }
 
-// Add adds a discovery to the list of managed discoveries
-func (dm *DiscoveryManager) Add(disc *discovery.PluggableDiscovery) error {
-	id := disc.GetID()
+// Start starts all the discoveries in this DiscoveryManager.
+// If the discoveries are already running, this function does nothing.
+func (dm *DiscoveryManager) Start() {
 	dm.discoveriesMutex.Lock()
 	defer dm.discoveriesMutex.Unlock()
-	if _, has := dm.discoveries[id]; has {
-		return errors.Errorf(tr("pluggable discovery already added: %s"), id)
+	if dm.discoveriesRunning {
+		return
 	}
-	dm.discoveries[id] = disc
-	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.discoveriesMutex.Lock()
-	d := dm.discoveries[id]
-	delete(dm.discoveries, id)
-	dm.discoveriesMutex.Unlock()
-	d.Quit()
-	logrus.Infof("Closed and removed discovery %s", id)
-}
+	go dm.feeder()
 
-// 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 {
 	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()
-			if err := f(d); err != nil {
-				errChan <- err
-			}
+			dm.startDiscovery(d)
+			wg.Done()
 		}(d)
 	}
+	wg.Wait()
+	dm.discoveriesRunning = true
+}
 
-	// Wait in a goroutine to collect eventual errors running a discovery.
-	// When all goroutines that are calling discoveries are done close the errors chan.
-	go func() {
-		wg.Wait()
-		close(errChan)
-	}()
+// Add adds a discovery to the list of managed discoveries
+func (dm *DiscoveryManager) Add(d *discovery.PluggableDiscovery) error {
+	dm.discoveriesMutex.Lock()
+	defer dm.discoveriesMutex.Unlock()
 
-	errs := []error{}
-	for err := range errChan {
-		errs = append(errs, err)
+	id := d.GetID()
+	if _, has := dm.discoveries[id]; has {
+		return errors.Errorf(tr("pluggable discovery already added: %s"), id)
 	}
-	return errs
-}
-
-// RunAll the discoveries for this DiscoveryManager,
-// returns an error for each discovery failing to run
-func (dm *DiscoveryManager) RunAll() []error {
-	return dm.parallelize(func(d *discovery.PluggableDiscovery) error {
-		if d.State() != discovery.Dead {
-			// This discovery is already alive, nothing to do
-			return nil
-		}
+	dm.discoveries[id] = d
 
-		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
-	})
+	if dm.discoveriesRunning {
+		dm.startDiscovery(d)
+	}
+	return nil
 }
 
-// StartAll the discoveries for this DiscoveryManager,
-// returns an error for each discovery failing to start
-func (dm *DiscoveryManager) StartAll() []error {
-	return dm.parallelize(func(d *discovery.PluggableDiscovery) error {
-		state := d.State()
-		if state != discovery.Idling {
-			// Already started
-			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
-	})
+// PortWatcher is a watcher for all discovery events (port connection/disconnection)
+type PortWatcher struct {
+	closeCB func()
+	feed    chan *discovery.Event
 }
 
-// StartSyncAll the discoveries for this DiscoveryManager,
-// returns an error for each discovery failing to start syncing
-func (dm *DiscoveryManager) StartSyncAll() (<-chan *discovery.Event, []error) {
-	eventSink := make(chan *discovery.Event, 5)
-	var wg sync.WaitGroup
-	errs := dm.parallelize(func(d *discovery.PluggableDiscovery) error {
-		state := d.State()
-		if state != discovery.Idling || state == discovery.Syncing {
-			// Already syncing
-			return nil
-		}
-
-		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)
-		}
+// Feed returns the feed of events coming from the discoveries
+func (pw *PortWatcher) Feed() <-chan *discovery.Event {
+	return pw.feed
+}
 
-		wg.Add(1)
-		go func() {
-			for ev := range eventCh {
-				eventSink <- ev
-			}
-			wg.Done()
-		}()
-		return nil
-	})
-	go func() {
-		wg.Wait()
-		eventSink <- &discovery.Event{Type: "quit"}
-		close(eventSink)
-	}()
-	return eventSink, errs
+// Close closes the PortWatcher
+func (pw *PortWatcher) Close() {
+	pw.closeCB()
 }
 
-// StopAll the discoveries for this DiscoveryManager,
-// returns an error for each discovery failing to stop
-func (dm *DiscoveryManager) StopAll() []error {
-	return dm.parallelize(func(d *discovery.PluggableDiscovery) error {
-		state := d.State()
-		if state != discovery.Syncing && state != discovery.Running {
-			// Not running nor syncing, nothing to stop
-			return nil
-		}
+// Watch starts a watcher for all discovery events (port connection/disconnection).
+// The watcher must be closed when it is no longer needed with the Close method.
+func (dm *DiscoveryManager) Watch() (*PortWatcher, error) {
+	dm.Start()
 
-		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
-	})
+	watcher := &PortWatcher{
+		feed: make(chan *discovery.Event),
+	}
+	watcher.closeCB = func() {
+		dm.watchersMutex.Lock()
+		delete(dm.watchers, watcher)
+		dm.watchersMutex.Unlock()
+		close(watcher.feed)
+	}
+	dm.watchersMutex.Lock()
+	dm.watchers[watcher] = true
+	dm.watchersMutex.Unlock()
+	return watcher, nil
 }
 
-// QuitAll quits all the discoveries managed by this DiscoveryManager.
-// Returns an error for each discovery that fails quitting
-func (dm *DiscoveryManager) QuitAll() []error {
-	errs := dm.parallelize(func(d *discovery.PluggableDiscovery) error {
-		if d.State() == discovery.Dead {
-			// Stop! Stop! It's already dead!
-			return nil
+func (dm *DiscoveryManager) startDiscovery(d *discovery.PluggableDiscovery) (discErr error) {
+	defer func() {
+		if discErr != nil {
+			logrus.Errorf("Discovery %s failed to run: %s", d.GetID(), discErr)
 		}
+	}()
 
-		d.Quit()
-		return nil
-	})
-	return errs
-}
-
-// List returns a list of available ports detected from all discoveries
-// and a list of errors for those discoveries that returned one.
-func (dm *DiscoveryManager) List() ([]*discovery.Port, []error) {
-	var wg sync.WaitGroup
-	// Use this struct to avoid the need of two separate
-	// channels for ports and errors.
-	type listMsg struct {
-		Err  error
-		Port *discovery.Port
+	if err := d.Run(); err != nil {
+		return fmt.Errorf(tr("discovery %[1]s process not started: %[2]w"), d.GetID(), err)
 	}
-	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()
-			if d.State() != discovery.Running {
-				// Discovery is not running, it won't return anything
-				return
-			}
-			ports, err := d.List()
-			if err != nil {
-				msgChan <- listMsg{Err: fmt.Errorf(tr("listing ports from discovery %[1]s: %[2]w"), d.GetID(), err)}
-			}
-			for _, p := range ports {
-				msgChan <- listMsg{Port: p}
-			}
-		}(d)
+	eventCh, err := d.StartSync(5)
+	if err != nil {
+		return fmt.Errorf("%s: %s", tr("starting discovery %s", d.GetID()), err)
 	}
 
 	go func() {
-		// Close the channel only after all goroutines are done
-		wg.Wait()
-		close(msgChan)
+		for ev := range eventCh {
+			dm.feed <- ev
+		}
 	}()
+	return nil
+}
 
-	ports := []*discovery.Port{}
-	errs := []error{}
-	for msg := range msgChan {
-		if msg.Err != nil {
-			errs = append(errs, msg.Err)
-		} else {
-			ports = append(ports, msg.Port)
+func (dm *DiscoveryManager) feeder() {
+	// Feed all watchers with data coming from the discoveries
+	for ev := range dm.feed {
+		dm.watchersMutex.Lock()
+		for watcher := range dm.watchers {
+			select {
+			case watcher.feed <- ev:
+				// OK
+			default:
+				// If the watcher is not able to process event fast enough
+				// remove the watcher from the list of watchers
+				go watcher.Close()
+			}
 		}
+		dm.cacheEvent(ev)
+		dm.watchersMutex.Unlock()
 	}
-	return ports, errs
 }
 
-// ListCachedPorts return the current list of ports detected from all discoveries
-func (dm *DiscoveryManager) ListCachedPorts() []*discovery.Port {
-	res := []*discovery.Port{}
+func (dm *DiscoveryManager) cacheEvent(ev *discovery.Event) {
+	// XXX: TODO
+}
+
+// List return the current list of ports detected from all discoveries
+func (dm *DiscoveryManager) List() []*discovery.Port {
+	dm.Start()
+
+	// XXX: Cache ports and return them
 	dm.discoveriesMutex.Lock()
-	discoveries := []*discovery.PluggableDiscovery{}
+	defer dm.discoveriesMutex.Unlock()
+	res := []*discovery.Port{}
 	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
-		}
 		res = append(res, d.ListCachedPorts()...)
 	}
 	return res
diff --git a/cli/arguments/port.go b/cli/arguments/port.go
index 12e7e28f5c7..d77215dcf58 100644
--- a/cli/arguments/port.go
+++ b/cli/arguments/port.go
@@ -106,31 +106,16 @@ func (p *Port) GetPort(instance *rpc.Instance, sk *sketch.Sketch) (*discovery.Po
 		return nil, errors.New("invalid instance")
 	}
 	dm := pm.DiscoveryManager()
-	if errs := dm.RunAll(); len(errs) == len(dm.IDs()) {
-		// All discoveries failed to run, we can't do anything
-		return nil, fmt.Errorf("%v", errs)
-	} else if len(errs) > 0 {
-		// If only some discoveries failed to run just tell the user and go on
-		for _, err := range errs {
-			feedback.Error(err)
-		}
-	}
-	eventChan, errs := dm.StartSyncAll()
-	if len(errs) > 0 {
-		return nil, fmt.Errorf("%v", errs)
+	watcher, err := dm.Watch()
+	if err != nil {
+		return nil, err
 	}
-
-	defer func() {
-		// Quit all discoveries at the end.
-		if errs := dm.QuitAll(); len(errs) > 0 {
-			logrus.Errorf("quitting discoveries when getting port metadata: %v", errs)
-		}
-	}()
+	defer watcher.Close()
 
 	deadline := time.After(p.timeout.Get())
 	for {
 		select {
-		case portEvent := <-eventChan:
+		case portEvent := <-watcher.Feed():
 			if portEvent.Type != "add" {
 				continue
 			}
diff --git a/cli/board/list.go b/cli/board/list.go
index 83f4e4418cd..b162e2a6f79 100644
--- a/cli/board/list.go
+++ b/cli/board/list.go
@@ -75,11 +75,12 @@ func runListCommand(cmd *cobra.Command, args []string) {
 }
 
 func watchList(cmd *cobra.Command, inst *rpc.Instance) {
-	eventsChan, err := board.Watch(inst.Id, nil)
+	eventsChan, closeCB, err := board.Watch(inst.Id)
 	if err != nil {
 		feedback.Errorf(tr("Error detecting boards: %v"), err)
 		os.Exit(errorcodes.ErrNetwork)
 	}
+	defer closeCB()
 
 	// This is done to avoid printing the header each time a new event is received
 	if feedback.GetFormat() == feedback.Text {
diff --git a/commands/board/list.go b/commands/board/list.go
index c2498889c10..84e7149b88f 100644
--- a/commands/board/list.go
+++ b/commands/board/list.go
@@ -16,6 +16,7 @@
 package board
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -183,22 +184,11 @@ func List(req *rpc.BoardListRequest) (r []*rpc.DetectedPort, e error) {
 	}
 
 	dm := pm.DiscoveryManager()
-	if errs := dm.RunAll(); len(errs) > 0 {
-		return nil, &arduino.UnavailableError{Message: tr("Error starting board discoveries"), Cause: fmt.Errorf("%v", errs)}
-	}
-	if errs := dm.StartAll(); len(errs) > 0 {
-		return nil, &arduino.UnavailableError{Message: tr("Error starting board discoveries"), Cause: fmt.Errorf("%v", errs)}
-	}
-	defer func() {
-		if errs := dm.StopAll(); len(errs) > 0 {
-			logrus.Error(errs)
-		}
-	}()
+	dm.Start()
 	time.Sleep(time.Duration(req.GetTimeout()) * time.Millisecond)
 
 	retVal := []*rpc.DetectedPort{}
-	ports, errs := pm.DiscoveryManager().List()
-	for _, port := range ports {
+	for _, port := range dm.List() {
 		boards, err := identify(pm, port)
 		if err != nil {
 			return nil, err
@@ -212,92 +202,49 @@ func List(req *rpc.BoardListRequest) (r []*rpc.DetectedPort, e error) {
 		}
 		retVal = append(retVal, b)
 	}
-	if len(errs) > 0 {
-		return retVal, &arduino.UnavailableError{Message: tr("Error getting board list"), Cause: fmt.Errorf("%v", errs)}
-	}
 	return retVal, nil
 }
 
 // Watch returns a channel that receives boards connection and disconnection events.
-// The discovery process can be interrupted by sending a message to the interrupt channel.
-func Watch(instanceID int32, interrupt <-chan bool) (<-chan *rpc.BoardListWatchResponse, error) {
+// It also returns a callback function that must be used to stop and dispose the watch.
+func Watch(instanceID int32) (<-chan *rpc.BoardListWatchResponse, func(), error) {
 	pm := commands.GetPackageManager(instanceID)
 	dm := pm.DiscoveryManager()
 
-	runErrs := dm.RunAll()
-	if len(runErrs) == len(dm.IDs()) {
-		// All discoveries failed to run, we can't do anything
-		return nil, &arduino.UnavailableError{Message: tr("Error starting board discoveries"), Cause: fmt.Errorf("%v", runErrs)}
+	watcher, err := dm.Watch()
+	if err != nil {
+		return nil, nil, err
 	}
 
-	eventsChan, errs := dm.StartSyncAll()
-	if len(runErrs) > 0 {
-		errs = append(runErrs, errs...)
-	}
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-ctx.Done()
+		watcher.Close()
+	}()
 
 	outChan := make(chan *rpc.BoardListWatchResponse)
-
 	go func() {
 		defer close(outChan)
-		for _, err := range errs {
-			outChan <- &rpc.BoardListWatchResponse{
-				EventType: "error",
-				Error:     err.Error(),
+		for event := range watcher.Feed() {
+			port := &rpc.DetectedPort{
+				Port: event.Port.ToRPC(),
 			}
-		}
-		for {
-			select {
-			case event := <-eventsChan:
-				if event.Type == "quit" {
-					// The discovery manager has closed its event channel because it's
-					// quitting all the discovery processes that are running, this
-					// means that the events channel we're listening from won't receive any
-					// more events.
-					// Handling this case is necessary when the board watcher is running and
-					// the instance being used is reinitialized since that quits all the
-					// discovery processes and reset the discovery manager. That would leave
-					// this goroutine listening forever on a "dead" channel and might even
-					// cause panics.
-					// This message avoid all this issues.
-					// It will be the client's task restarting the board watcher if necessary,
-					// this host won't attempt restarting it.
-					outChan <- &rpc.BoardListWatchResponse{
-						EventType: event.Type,
-					}
-					return
-				}
-
-				port := &rpc.DetectedPort{
-					Port: event.Port.ToRPC(),
-				}
 
-				boardsError := ""
-				if event.Type == "add" {
-					boards, err := identify(pm, event.Port)
-					if err != nil {
-						boardsError = err.Error()
-					}
-					port.MatchingBoards = boards
+			boardsError := ""
+			if event.Type == "add" {
+				boards, err := identify(pm, event.Port)
+				if err != nil {
+					boardsError = err.Error()
 				}
-				outChan <- &rpc.BoardListWatchResponse{
-					EventType: event.Type,
-					Port:      port,
-					Error:     boardsError,
-				}
-			case <-interrupt:
-				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", err),
-					}
-				}
-				return
+				port.MatchingBoards = boards
+			}
+			outChan <- &rpc.BoardListWatchResponse{
+				EventType: event.Type,
+				Port:      port,
+				Error:     boardsError,
 			}
 		}
 	}()
 
-	return outChan, nil
+	return outChan, cancel, nil
 }
diff --git a/commands/daemon/daemon.go b/commands/daemon/daemon.go
index a82656bbe70..03045c87a78 100644
--- a/commands/daemon/daemon.go
+++ b/commands/daemon/daemon.go
@@ -36,9 +36,7 @@ import (
 	"github.com/arduino/arduino-cli/i18n"
 	rpc "github.com/arduino/arduino-cli/rpc/cc/arduino/cli/commands/v1"
 	"github.com/sirupsen/logrus"
-	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
-	"google.golang.org/grpc/status"
 )
 
 // ArduinoCoreServerImpl FIXMEDOC
@@ -109,42 +107,35 @@ func (s *ArduinoCoreServerImpl) BoardListWatch(stream rpc.ArduinoCoreService_Boa
 		return err
 	}
 
-	interrupt := make(chan bool, 1)
+	eventsChan, closeWatcher, err := board.Watch(msg.Instance.Id)
+	if err != nil {
+		return convertErrorToRPCStatus(err)
+	}
+
 	go func() {
-		defer close(interrupt)
+		defer closeWatcher()
 		for {
 			msg, err := stream.Recv()
 			// Handle client closing the stream and eventual errors
 			if err == io.EOF {
 				logrus.Info("boards watcher stream closed")
-				interrupt <- true
-				return
-			} else if st, ok := status.FromError(err); ok && st.Code() == codes.Canceled {
-				logrus.Info("boards watcher interrupted by host")
 				return
-			} else if err != nil {
+			}
+			if err != nil {
 				logrus.Infof("interrupting boards watcher: %v", err)
-				interrupt <- true
 				return
 			}
 
 			// Message received, does the client want to interrupt?
 			if msg != nil && msg.Interrupt {
 				logrus.Info("boards watcher interrupted by client")
-				interrupt <- msg.Interrupt
 				return
 			}
 		}
 	}()
 
-	eventsChan, err := board.Watch(msg.Instance.Id, interrupt)
-	if err != nil {
-		return convertErrorToRPCStatus(err)
-	}
-
 	for event := range eventsChan {
-		err = stream.Send(event)
-		if err != nil {
+		if err := stream.Send(event); err != nil {
 			logrus.Infof("sending board watch message: %v", err)
 		}
 	}

From dcf465e797be654ea2942c7dce083706b75b548a Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Wed, 13 Jul 2022 16:14:28 +0200
Subject: [PATCH 02/15] Added discovery id in discovery.Event struct

---
 arduino/discovery/discovery.go | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/arduino/discovery/discovery.go b/arduino/discovery/discovery.go
index 40067cc3e43..37c7f750c0e 100644
--- a/arduino/discovery/discovery.go
+++ b/arduino/discovery/discovery.go
@@ -121,8 +121,9 @@ func (p *Port) String() string {
 
 // Event is a pluggable discovery event
 type Event struct {
-	Type string
-	Port *Port
+	Type        string
+	Port        *Port
+	DiscoveryID string
 }
 
 // New create and connect to the given pluggable discovery
@@ -178,7 +179,7 @@ func (disc *PluggableDiscovery) jsonDecodeLoop(in io.Reader, outChan chan<- *dis
 			disc.statusMutex.Lock()
 			disc.cachedPorts[msg.Port.Address+"|"+msg.Port.Protocol] = msg.Port
 			if disc.eventChan != nil {
-				disc.eventChan <- &Event{"add", msg.Port}
+				disc.eventChan <- &Event{"add", msg.Port, disc.GetID()}
 			}
 			disc.statusMutex.Unlock()
 		} else if msg.EventType == "remove" {
@@ -189,7 +190,7 @@ func (disc *PluggableDiscovery) jsonDecodeLoop(in io.Reader, outChan chan<- *dis
 			disc.statusMutex.Lock()
 			delete(disc.cachedPorts, msg.Port.Address+"|"+msg.Port.Protocol)
 			if disc.eventChan != nil {
-				disc.eventChan <- &Event{"remove", msg.Port}
+				disc.eventChan <- &Event{"remove", msg.Port, disc.GetID()}
 			}
 			disc.statusMutex.Unlock()
 		} else {

From 81dc23f45266c5a6380386d49d7532d107a084b6 Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Wed, 13 Jul 2022 16:15:52 +0200
Subject: [PATCH 03/15] Cache active ports and transmit them when a new watcher
 connects

---
 .../discoverymanager/discoverymanager.go      | 49 ++++++++++++++-----
 1 file changed, 37 insertions(+), 12 deletions(-)

diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index e6857e77de0..f5036502101 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -34,6 +34,7 @@ type DiscoveryManager struct {
 	feed               chan *discovery.Event
 	watchersMutex      sync.Mutex
 	watchers           map[*PortWatcher]bool
+	watchersCache      map[string]map[string]*discovery.Event
 }
 
 var tr = i18n.Tr
@@ -41,9 +42,10 @@ var tr = i18n.Tr
 // New creates a new DiscoveryManager
 func New() *DiscoveryManager {
 	return &DiscoveryManager{
-		discoveries: map[string]*discovery.PluggableDiscovery{},
-		watchers:    map[*PortWatcher]bool{},
-		feed:        make(chan *discovery.Event, 50),
+		discoveries:   map[string]*discovery.PluggableDiscovery{},
+		watchers:      map[*PortWatcher]bool{},
+		feed:          make(chan *discovery.Event, 50),
+		watchersCache: map[string]map[string]*discovery.Event{},
 	}
 }
 
@@ -139,9 +141,16 @@ func (dm *DiscoveryManager) Watch() (*PortWatcher, error) {
 		dm.watchersMutex.Unlock()
 		close(watcher.feed)
 	}
-	dm.watchersMutex.Lock()
-	dm.watchers[watcher] = true
-	dm.watchersMutex.Unlock()
+	go func() {
+		dm.watchersMutex.Lock()
+		for _, cache := range dm.watchersCache {
+			for _, ev := range cache {
+				watcher.feed <- ev
+			}
+		}
+		dm.watchers[watcher] = true
+		dm.watchersMutex.Unlock()
+	}()
 	return watcher, nil
 }
 
@@ -188,19 +197,35 @@ func (dm *DiscoveryManager) feeder() {
 }
 
 func (dm *DiscoveryManager) cacheEvent(ev *discovery.Event) {
-	// XXX: TODO
+	cache := dm.watchersCache[ev.DiscoveryID]
+	if cache == nil {
+		cache = map[string]*discovery.Event{}
+		dm.watchersCache[ev.DiscoveryID] = cache
+	}
+
+	eventID := ev.Port.Address + "|" + ev.Port.Protocol
+	switch ev.Type {
+	case "add":
+		cache[eventID] = ev
+	case "remove":
+		delete(cache, eventID)
+	default:
+		logrus.Errorf("Unhandled event from discovery: %s", ev.Type)
+		return
+	}
 }
 
 // List return the current list of ports detected from all discoveries
 func (dm *DiscoveryManager) List() []*discovery.Port {
 	dm.Start()
 
-	// XXX: Cache ports and return them
-	dm.discoveriesMutex.Lock()
-	defer dm.discoveriesMutex.Unlock()
 	res := []*discovery.Port{}
-	for _, d := range dm.discoveries {
-		res = append(res, d.ListCachedPorts()...)
+	dm.watchersMutex.Lock()
+	defer dm.watchersMutex.Unlock()
+	for _, cache := range dm.watchersCache {
+		for _, ev := range cache {
+			res = append(res, ev.Port)
+		}
 	}
 	return res
 }

From 599fd0c17edb005812f8474a210dad1fc2e1aa78 Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Wed, 13 Jul 2022 23:48:06 +0200
Subject: [PATCH 04/15] Correctly handle discovery cleanup

---
 arduino/discovery/discovery.go                | 24 +++++++++----------
 .../discoverymanager/discoverymanager.go      | 14 +++++++----
 2 files changed, 21 insertions(+), 17 deletions(-)

diff --git a/arduino/discovery/discovery.go b/arduino/discovery/discovery.go
index 37c7f750c0e..4f77a6290ed 100644
--- a/arduino/discovery/discovery.go
+++ b/arduino/discovery/discovery.go
@@ -277,10 +277,7 @@ func (disc *PluggableDiscovery) killProcess() error {
 	}
 	disc.statusMutex.Lock()
 	defer disc.statusMutex.Unlock()
-	if disc.eventChan != nil {
-		close(disc.eventChan)
-		disc.eventChan = nil
-	}
+	disc.stopSync()
 	disc.state = Dead
 	logrus.Infof("killed discovery %s process", disc.id)
 	return nil
@@ -367,13 +364,18 @@ func (disc *PluggableDiscovery) Stop() error {
 	}
 	disc.statusMutex.Lock()
 	defer disc.statusMutex.Unlock()
-	disc.cachedPorts = map[string]*Port{}
+	disc.stopSync()
+	disc.state = Idling
+	return nil
+}
+
+func (disc *PluggableDiscovery) stopSync() {
 	if disc.eventChan != nil {
+		disc.eventChan <- &Event{"stop", nil, disc.GetID()}
 		close(disc.eventChan)
 		disc.eventChan = nil
+		disc.cachedPorts = map[string]*Port{}
 	}
-	disc.state = Idling
-	return nil
 }
 
 // Quit terminates the discovery. No more commands can be accepted by the discovery.
@@ -427,12 +429,8 @@ func (disc *PluggableDiscovery) StartSync(size int) (<-chan *Event, error) {
 	disc.statusMutex.Lock()
 	defer disc.statusMutex.Unlock()
 	disc.state = Syncing
-	disc.cachedPorts = map[string]*Port{}
-	if disc.eventChan != nil {
-		// In case there is already an existing event channel in use we close it
-		// before creating a new one.
-		close(disc.eventChan)
-	}
+	// In case there is already an existing event channel in use we close it before creating a new one.
+	disc.stopSync()
 	c := make(chan *Event, size)
 	disc.eventChan = c
 	return c, nil
diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index f5036502101..8636654d5aa 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -52,12 +52,15 @@ func New() *DiscoveryManager {
 // Clear resets the DiscoveryManager to its initial state
 func (dm *DiscoveryManager) Clear() {
 	dm.discoveriesMutex.Lock()
-	for _, d := range dm.discoveries {
-		d.Quit()
-		logrus.Infof("Closed and removed discovery %s", d.GetID())
+	defer dm.discoveriesMutex.Unlock()
+
+	if dm.discoveriesRunning {
+		for _, d := range dm.discoveries {
+			d.Quit()
+			logrus.Infof("Closed and removed discovery %s", d.GetID())
+		}
 	}
 	dm.discoveries = map[string]*discovery.PluggableDiscovery{}
-	dm.discoveriesMutex.Unlock()
 }
 
 // IDs returns the list of discoveries' ids in this DiscoveryManager
@@ -209,6 +212,9 @@ func (dm *DiscoveryManager) cacheEvent(ev *discovery.Event) {
 		cache[eventID] = ev
 	case "remove":
 		delete(cache, eventID)
+	case "quit":
+		// Remove all the events for this discovery
+		delete(dm.watchersCache, ev.DiscoveryID)
 	default:
 		logrus.Errorf("Unhandled event from discovery: %s", ev.Type)
 		return

From 804726ba3b0f4606fb9ae4eb0ca0b145f5578883 Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Thu, 14 Jul 2022 00:10:50 +0200
Subject: [PATCH 05/15] Fixed wrong test

---
 .../cores/packagemanager/package_manager_test.go   | 14 ++++++--------
 1 file changed, 6 insertions(+), 8 deletions(-)

diff --git a/arduino/cores/packagemanager/package_manager_test.go b/arduino/cores/packagemanager/package_manager_test.go
index ec12a6070b9..9cb1509dca0 100644
--- a/arduino/cores/packagemanager/package_manager_test.go
+++ b/arduino/cores/packagemanager/package_manager_test.go
@@ -329,16 +329,14 @@ func TestPackageManagerClear(t *testing.T) {
 	packageManager := packagemanager.NewPackageManager(customHardware, customHardware, customHardware, customHardware, "test")
 	packageManager.LoadHardwareFromDirectory(customHardware)
 
-	// Creates another PackageManager but don't load the hardware
-	emptyPackageManager := packagemanager.NewPackageManager(customHardware, customHardware, customHardware, customHardware, "test")
+	// Check that the hardware is loaded
+	require.NotEmpty(t, packageManager.Packages)
 
-	// Verifies they're not equal
-	require.NotEqual(t, packageManager, emptyPackageManager)
-
-	// Clear the first PackageManager that contains loaded hardware
+	// Clear the package manager
 	packageManager.Clear()
-	// Verifies both PackageManagers are now equal
-	require.Equal(t, packageManager, emptyPackageManager)
+
+	// Check that the hardware is cleared
+	require.Empty(t, packageManager.Packages)
 }
 
 func TestFindToolsRequiredFromPlatformRelease(t *testing.T) {

From f638a97dc107e627629b6de98c0052780da092ca Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Fri, 15 Jul 2022 01:32:34 +0200
Subject: [PATCH 06/15] Correctly handle discovery cleanup and re-add

---
 arduino/discovery/discovery.go                |  5 +-
 .../discoverymanager/discoverymanager.go      | 55 ++++++++++---------
 2 files changed, 34 insertions(+), 26 deletions(-)

diff --git a/arduino/discovery/discovery.go b/arduino/discovery/discovery.go
index 4f77a6290ed..ead70adecc6 100644
--- a/arduino/discovery/discovery.go
+++ b/arduino/discovery/discovery.go
@@ -371,10 +371,13 @@ func (disc *PluggableDiscovery) Stop() error {
 
 func (disc *PluggableDiscovery) stopSync() {
 	if disc.eventChan != nil {
+		for _, port := range disc.cachedPorts {
+			disc.eventChan <- &Event{"remove", port, disc.GetID()}
+		}
+		disc.cachedPorts = map[string]*Port{}
 		disc.eventChan <- &Event{"stop", nil, disc.GetID()}
 		close(disc.eventChan)
 		disc.eventChan = nil
-		disc.cachedPorts = map[string]*Port{}
 	}
 }
 
diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index 8636654d5aa..bc418f215c4 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -18,6 +18,7 @@ package discoverymanager
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/arduino/arduino-cli/arduino/discovery"
 	"github.com/arduino/arduino-cli/i18n"
@@ -83,7 +84,12 @@ func (dm *DiscoveryManager) Start() {
 		return
 	}
 
-	go dm.feeder()
+	go func() {
+		// Feed all watchers with data coming from the discoveries
+		for ev := range dm.feed {
+			dm.feedEvent(ev)
+		}
+	}()
 
 	var wg sync.WaitGroup
 	for _, d := range dm.discoveries {
@@ -136,13 +142,13 @@ func (dm *DiscoveryManager) Watch() (*PortWatcher, error) {
 	dm.Start()
 
 	watcher := &PortWatcher{
-		feed: make(chan *discovery.Event),
+		feed: make(chan *discovery.Event, 10),
 	}
 	watcher.closeCB = func() {
 		dm.watchersMutex.Lock()
 		delete(dm.watchers, watcher)
-		dm.watchersMutex.Unlock()
 		close(watcher.feed)
+		dm.watchersMutex.Unlock()
 	}
 	go func() {
 		dm.watchersMutex.Lock()
@@ -180,44 +186,43 @@ func (dm *DiscoveryManager) startDiscovery(d *discovery.PluggableDiscovery) (dis
 	return nil
 }
 
-func (dm *DiscoveryManager) feeder() {
-	// Feed all watchers with data coming from the discoveries
-	for ev := range dm.feed {
-		dm.watchersMutex.Lock()
-		for watcher := range dm.watchers {
-			select {
-			case watcher.feed <- ev:
-				// OK
-			default:
-				// If the watcher is not able to process event fast enough
-				// remove the watcher from the list of watchers
-				go watcher.Close()
-			}
+func (dm *DiscoveryManager) feedEvent(ev *discovery.Event) {
+	dm.watchersMutex.Lock()
+	defer dm.watchersMutex.Unlock()
+
+	if ev.Type == "stop" {
+		// Remove all the cached events for the terminating discovery
+		delete(dm.watchersCache, ev.DiscoveryID)
+		return
+	}
+
+	// Send the event to all watchers
+	for watcher := range dm.watchers {
+		select {
+		case watcher.feed <- ev:
+			// OK
+		case <-time.After(time.Millisecond * 500):
+			// If the watcher is not able to process event fast enough
+			// remove the watcher from the list of watchers
+			logrus.Info("Watcher is not able to process events fast enough, removing it from the list of watchers")
+			delete(dm.watchers, watcher)
 		}
-		dm.cacheEvent(ev)
-		dm.watchersMutex.Unlock()
 	}
-}
 
-func (dm *DiscoveryManager) cacheEvent(ev *discovery.Event) {
+	// Cache the event for the discovery
 	cache := dm.watchersCache[ev.DiscoveryID]
 	if cache == nil {
 		cache = map[string]*discovery.Event{}
 		dm.watchersCache[ev.DiscoveryID] = cache
 	}
-
 	eventID := ev.Port.Address + "|" + ev.Port.Protocol
 	switch ev.Type {
 	case "add":
 		cache[eventID] = ev
 	case "remove":
 		delete(cache, eventID)
-	case "quit":
-		// Remove all the events for this discovery
-		delete(dm.watchersCache, ev.DiscoveryID)
 	default:
 		logrus.Errorf("Unhandled event from discovery: %s", ev.Type)
-		return
 	}
 }
 

From 601c243b611738e709fbb898e0589d67ab875be2 Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Wed, 20 Jul 2022 16:35:20 +0200
Subject: [PATCH 07/15] Added some doc comments in the source code

---
 arduino/discovery/discovery.go                |  2 ++
 .../discoverymanager/discoverymanager.go      | 24 ++++++++++++-------
 2 files changed, 18 insertions(+), 8 deletions(-)

diff --git a/arduino/discovery/discovery.go b/arduino/discovery/discovery.go
index ead70adecc6..62b813f5f02 100644
--- a/arduino/discovery/discovery.go
+++ b/arduino/discovery/discovery.go
@@ -371,6 +371,8 @@ func (disc *PluggableDiscovery) Stop() error {
 
 func (disc *PluggableDiscovery) stopSync() {
 	if disc.eventChan != nil {
+		// When stopping sync send a batch of "remove" events for
+		// all the active ports.
 		for _, port := range disc.cachedPorts {
 			disc.eventChan <- &Event{"remove", port, disc.GetID()}
 		}
diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index bc418f215c4..381f2d11207 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -26,16 +26,20 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-// DiscoveryManager is required to handle multiple pluggable-discovery that
-// may be shared across platforms
+// DiscoveryManager manages the many-to-many communication between all pluggable
+// discoveries and all watchers. Each PluggableDiscovery, once started, will
+// produce a sequence of "events". These events will be broadcasted to all
+// listening Watcher.
+// The DiscoveryManager will not start the discoveries until the Start method
+// is called.
 type DiscoveryManager struct {
 	discoveriesMutex   sync.Mutex
-	discoveries        map[string]*discovery.PluggableDiscovery
-	discoveriesRunning bool
-	feed               chan *discovery.Event
+	discoveries        map[string]*discovery.PluggableDiscovery // all registered PluggableDiscovery
+	discoveriesRunning bool                                     // set to true once discoveries are started
+	feed               chan *discovery.Event                    // all events will pass through this channel
 	watchersMutex      sync.Mutex
-	watchers           map[*PortWatcher]bool
-	watchersCache      map[string]map[string]*discovery.Event
+	watchers           map[*PortWatcher]bool                  // all registered Watcher
+	watchersCache      map[string]map[string]*discovery.Event // this is a cache of all active ports
 }
 
 var tr = i18n.Tr
@@ -85,7 +89,7 @@ func (dm *DiscoveryManager) Start() {
 	}
 
 	go func() {
-		// Feed all watchers with data coming from the discoveries
+		// Send all events coming from the feed channel to all active watchers
 		for ev := range dm.feed {
 			dm.feedEvent(ev)
 		}
@@ -152,11 +156,13 @@ func (dm *DiscoveryManager) Watch() (*PortWatcher, error) {
 	}
 	go func() {
 		dm.watchersMutex.Lock()
+		// When a watcher is started, send all the current active ports first...
 		for _, cache := range dm.watchersCache {
 			for _, ev := range cache {
 				watcher.feed <- ev
 			}
 		}
+		// ...and after that add the watcher to the list of watchers receiving events
 		dm.watchers[watcher] = true
 		dm.watchersMutex.Unlock()
 	}()
@@ -165,6 +171,7 @@ func (dm *DiscoveryManager) Watch() (*PortWatcher, error) {
 
 func (dm *DiscoveryManager) startDiscovery(d *discovery.PluggableDiscovery) (discErr error) {
 	defer func() {
+		// If this function returns an error log it
 		if discErr != nil {
 			logrus.Errorf("Discovery %s failed to run: %s", d.GetID(), discErr)
 		}
@@ -179,6 +186,7 @@ func (dm *DiscoveryManager) startDiscovery(d *discovery.PluggableDiscovery) (dis
 	}
 
 	go func() {
+		// Transfer all incoming events from this discovery to the feed channel
 		for ev := range eventCh {
 			dm.feed <- ev
 		}

From 081551b8d530d0ee35659e7cd2ad0dc8ce7a722f Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Wed, 20 Jul 2022 16:41:22 +0200
Subject: [PATCH 08/15] Move Unlock under defer

---
 arduino/discovery/discoverymanager/discoverymanager.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index 381f2d11207..54c9ce77176 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -150,9 +150,9 @@ func (dm *DiscoveryManager) Watch() (*PortWatcher, error) {
 	}
 	watcher.closeCB = func() {
 		dm.watchersMutex.Lock()
+		defer dm.watchersMutex.Unlock()
 		delete(dm.watchers, watcher)
 		close(watcher.feed)
-		dm.watchersMutex.Unlock()
 	}
 	go func() {
 		dm.watchersMutex.Lock()

From 6c1c8dfdd06f3248abc336d20d71e9e07519a088 Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Thu, 4 Aug 2022 14:26:28 +0200
Subject: [PATCH 09/15] Factored subrotuine into a function

it will be useful in the next commits.
---
 .../discoverymanager/discoverymanager.go      | 28 +++++++++++--------
 1 file changed, 16 insertions(+), 12 deletions(-)

diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index 54c9ce77176..3ef6d3f58aa 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -198,24 +198,28 @@ func (dm *DiscoveryManager) feedEvent(ev *discovery.Event) {
 	dm.watchersMutex.Lock()
 	defer dm.watchersMutex.Unlock()
 
+	sendToAllWatchers := func(ev *discovery.Event) {
+		// Send the event to all watchers
+		for watcher := range dm.watchers {
+			select {
+			case watcher.feed <- ev:
+				// OK
+			case <-time.After(time.Millisecond * 500):
+				// If the watcher is not able to process event fast enough
+				// remove the watcher from the list of watchers
+				logrus.Info("Watcher is not able to process events fast enough, removing it from the list of watchers")
+				delete(dm.watchers, watcher)
+			}
+		}
+	}
+
 	if ev.Type == "stop" {
 		// Remove all the cached events for the terminating discovery
 		delete(dm.watchersCache, ev.DiscoveryID)
 		return
 	}
 
-	// Send the event to all watchers
-	for watcher := range dm.watchers {
-		select {
-		case watcher.feed <- ev:
-			// OK
-		case <-time.After(time.Millisecond * 500):
-			// If the watcher is not able to process event fast enough
-			// remove the watcher from the list of watchers
-			logrus.Info("Watcher is not able to process events fast enough, removing it from the list of watchers")
-			delete(dm.watchers, watcher)
-		}
-	}
+	sendToAllWatchers(ev)
 
 	// Cache the event for the discovery
 	cache := dm.watchersCache[ev.DiscoveryID]

From aeec293d25d8f5830f551db67af41b856bbc5ff7 Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Thu, 4 Aug 2022 14:34:18 +0200
Subject: [PATCH 10/15] Do not cache ports in the DiscoveryClient

there is already a cache in the DiscoveryManager there is no need to
duplicate it.
---
 arduino/discovery/discovery.go                | 23 -------------------
 .../discoverymanager/discoverymanager.go      | 17 +++++++++++++-
 2 files changed, 16 insertions(+), 24 deletions(-)

diff --git a/arduino/discovery/discovery.go b/arduino/discovery/discovery.go
index 62b813f5f02..64e45ec7424 100644
--- a/arduino/discovery/discovery.go
+++ b/arduino/discovery/discovery.go
@@ -57,7 +57,6 @@ type PluggableDiscovery struct {
 	incomingMessagesError error
 	state                 int
 	eventChan             chan<- *Event
-	cachedPorts           map[string]*Port
 }
 
 type discoveryMessage struct {
@@ -132,7 +131,6 @@ func New(id string, args ...string) *PluggableDiscovery {
 		id:          id,
 		processArgs: args,
 		state:       Dead,
-		cachedPorts: map[string]*Port{},
 	}
 }
 
@@ -177,7 +175,6 @@ func (disc *PluggableDiscovery) jsonDecodeLoop(in io.Reader, outChan chan<- *dis
 				return
 			}
 			disc.statusMutex.Lock()
-			disc.cachedPorts[msg.Port.Address+"|"+msg.Port.Protocol] = msg.Port
 			if disc.eventChan != nil {
 				disc.eventChan <- &Event{"add", msg.Port, disc.GetID()}
 			}
@@ -188,7 +185,6 @@ func (disc *PluggableDiscovery) jsonDecodeLoop(in io.Reader, outChan chan<- *dis
 				return
 			}
 			disc.statusMutex.Lock()
-			delete(disc.cachedPorts, msg.Port.Address+"|"+msg.Port.Protocol)
 			if disc.eventChan != nil {
 				disc.eventChan <- &Event{"remove", msg.Port, disc.GetID()}
 			}
@@ -371,12 +367,6 @@ func (disc *PluggableDiscovery) Stop() error {
 
 func (disc *PluggableDiscovery) stopSync() {
 	if disc.eventChan != nil {
-		// When stopping sync send a batch of "remove" events for
-		// all the active ports.
-		for _, port := range disc.cachedPorts {
-			disc.eventChan <- &Event{"remove", port, disc.GetID()}
-		}
-		disc.cachedPorts = map[string]*Port{}
 		disc.eventChan <- &Event{"stop", nil, disc.GetID()}
 		close(disc.eventChan)
 		disc.eventChan = nil
@@ -440,16 +430,3 @@ func (disc *PluggableDiscovery) StartSync(size int) (<-chan *Event, error) {
 	disc.eventChan = c
 	return c, nil
 }
-
-// ListCachedPorts returns a list of the available ports. The list is a cache of all the
-// add/remove events happened from the StartSync call and it will not consume any
-// resource from the underliying discovery.
-func (disc *PluggableDiscovery) ListCachedPorts() []*Port {
-	disc.statusMutex.Lock()
-	defer disc.statusMutex.Unlock()
-	res := []*Port{}
-	for _, port := range disc.cachedPorts {
-		res = append(res, port)
-	}
-	return res
-}
diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index 3ef6d3f58aa..ccbd433038e 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -214,7 +214,22 @@ func (dm *DiscoveryManager) feedEvent(ev *discovery.Event) {
 	}
 
 	if ev.Type == "stop" {
-		// Remove all the cached events for the terminating discovery
+		// Send remove events for all the cached ports of the terminating discovery
+		cache := dm.watchersCache[ev.DiscoveryID]
+		for _, addEv := range cache {
+			removeEv := &discovery.Event{
+				Type: "remove",
+				Port: &discovery.Port{
+					Address:       addEv.Port.Address,
+					AddressLabel:  addEv.Port.AddressLabel,
+					Protocol:      addEv.Port.Protocol,
+					ProtocolLabel: addEv.Port.ProtocolLabel},
+				DiscoveryID: addEv.DiscoveryID,
+			}
+			sendToAllWatchers(removeEv)
+		}
+
+		// Remove the cache for the terminating discovery
 		delete(dm.watchersCache, ev.DiscoveryID)
 		return
 	}

From 7f68a75d73339fd7233092bbb4c26b29cab2ef9a Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Mon, 8 Aug 2022 17:20:01 +0200
Subject: [PATCH 11/15] Discovery: eventChan must be protected by mutex when
 doing START_SYNC

otherwise the discovery may send some events before the eventChan is
setup (and those events will be lost)
---
 arduino/discovery/discovery.go | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/arduino/discovery/discovery.go b/arduino/discovery/discovery.go
index 64e45ec7424..9def18596e7 100644
--- a/arduino/discovery/discovery.go
+++ b/arduino/discovery/discovery.go
@@ -407,6 +407,9 @@ func (disc *PluggableDiscovery) List() ([]*Port, error) {
 // The event channel must be consumed as quickly as possible since it may block the
 // discovery if it becomes full. The channel size is configurable.
 func (disc *PluggableDiscovery) StartSync(size int) (<-chan *Event, error) {
+	disc.statusMutex.Lock()
+	defer disc.statusMutex.Unlock()
+
 	if err := disc.sendCommand("START_SYNC\n"); err != nil {
 		return nil, err
 	}
@@ -421,8 +424,6 @@ func (disc *PluggableDiscovery) StartSync(size int) (<-chan *Event, error) {
 		return nil, errors.Errorf(tr("communication out of sync, expected '%[1]s', received '%[2]s'"), "OK", msg.Message)
 	}
 
-	disc.statusMutex.Lock()
-	defer disc.statusMutex.Unlock()
 	disc.state = Syncing
 	// In case there is already an existing event channel in use we close it before creating a new one.
 	disc.stopSync()

From 1f14c1792e567a4fe73f7d8bc05844eca191658e Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Mon, 8 Aug 2022 17:21:40 +0200
Subject: [PATCH 12/15] Increased error level for logging watchers that lags

---
 arduino/discovery/discoverymanager/discoverymanager.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index ccbd433038e..40928544948 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -207,7 +207,7 @@ func (dm *DiscoveryManager) feedEvent(ev *discovery.Event) {
 			case <-time.After(time.Millisecond * 500):
 				// If the watcher is not able to process event fast enough
 				// remove the watcher from the list of watchers
-				logrus.Info("Watcher is not able to process events fast enough, removing it from the list of watchers")
+				logrus.Error("Watcher is not able to process events fast enough, removing it from the list of watchers")
 				delete(dm.watchers, watcher)
 			}
 		}

From baf517da6580ba55b36c44e893e36f800706ca4b Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Mon, 8 Aug 2022 20:00:51 +0200
Subject: [PATCH 13/15] Updated discvoery_client to the latest API

---
 arduino/discovery/discovery_client/go.mod  |  2 +-
 arduino/discovery/discovery_client/main.go | 73 ++++++++++------------
 2 files changed, 34 insertions(+), 41 deletions(-)

diff --git a/arduino/discovery/discovery_client/go.mod b/arduino/discovery/discovery_client/go.mod
index 5fcd8c8e4da..45d45abf132 100644
--- a/arduino/discovery/discovery_client/go.mod
+++ b/arduino/discovery/discovery_client/go.mod
@@ -7,6 +7,7 @@ replace github.com/arduino/arduino-cli => ../../..
 require (
 	github.com/arduino/arduino-cli v0.0.0-00010101000000-000000000000
 	github.com/gizak/termui/v3 v3.1.0
+	github.com/sirupsen/logrus v1.4.2
 )
 
 require (
@@ -20,7 +21,6 @@ require (
 	github.com/nsf/termbox-go v0.0.0-20190121233118-02980233997d // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
-	github.com/sirupsen/logrus v1.4.2 // indirect
 	golang.org/x/net v0.0.0-20210505024714-0287a6fb4125 // indirect
 	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
 	golang.org/x/text v0.3.6 // indirect
diff --git a/arduino/discovery/discovery_client/main.go b/arduino/discovery/discovery_client/main.go
index 2f033a604a3..8306da7462e 100644
--- a/arduino/discovery/discovery_client/main.go
+++ b/arduino/discovery/discovery_client/main.go
@@ -21,36 +21,28 @@ import (
 	"log"
 	"os"
 	"sort"
-	"time"
 
 	"github.com/arduino/arduino-cli/arduino/discovery"
+	"github.com/arduino/arduino-cli/arduino/discovery/discoverymanager"
 	ui "github.com/gizak/termui/v3"
 	"github.com/gizak/termui/v3/widgets"
+	"github.com/sirupsen/logrus"
 )
 
 func main() {
-	discoveries := []*discovery.PluggableDiscovery{}
-	discEvent := make(chan *discovery.Event)
+	logrus.SetLevel(logrus.ErrorLevel)
+	dm := discoverymanager.New()
 	for _, discCmd := range os.Args[1:] {
-		disc := discovery.New("", discCmd)
-		if err := disc.Run(); err != nil {
-			log.Fatal("Error starting discovery:", err)
-		}
-		if err := disc.Start(); err != nil {
-			log.Fatal("Error starting discovery:", err)
-		}
-		eventChan, err := disc.StartSync(10)
-		if err != nil {
-			log.Fatal("Error starting discovery:", err)
-		}
-		go func() {
-			for msg := range eventChan {
-				discEvent <- msg
-			}
-		}()
-		discoveries = append(discoveries, disc)
+		disc := discovery.New(discCmd, discCmd)
+		dm.Add(disc)
 	}
+	dm.Start()
 
+	activePorts := map[string]*discovery.Port{}
+	watcher, err := dm.Watch()
+	if err != nil {
+		log.Fatalf("failed to start discvoeries: %v", err)
+	}
 	if err := ui.Init(); err != nil {
 		log.Fatalf("failed to initialize termui: %v", err)
 	}
@@ -66,15 +58,20 @@ func main() {
 	updateList := func() {
 		rows := []string{}
 		rows = append(rows, "Available ports list:")
-		for _, disc := range discoveries {
-			for i, port := range disc.ListCachedPorts() {
-				rows = append(rows, fmt.Sprintf(" [%04d] Address: %s", i, port.AddressLabel))
-				rows = append(rows, fmt.Sprintf("        Protocol: %s", port.ProtocolLabel))
-				keys := port.Properties.Keys()
-				sort.Strings(keys)
-				for _, k := range keys {
-					rows = append(rows, fmt.Sprintf("                  %s=%s", k, port.Properties.Get(k)))
-				}
+
+		ids := sort.StringSlice{}
+		for id := range activePorts {
+			ids = append(ids, id)
+		}
+		ids.Sort()
+		for _, id := range ids {
+			port := activePorts[id]
+			rows = append(rows, fmt.Sprintf("> Address: %s", port.AddressLabel))
+			rows = append(rows, fmt.Sprintf("  Protocol: %s", port.ProtocolLabel))
+			keys := port.Properties.Keys()
+			sort.Strings(keys)
+			for _, k := range keys {
+				rows = append(rows, fmt.Sprintf("                  %s=%s", k, port.Properties.Get(k)))
 			}
 		}
 		l.Rows = rows
@@ -123,20 +120,16 @@ out:
 				previousKey = e.ID
 			}
 
-		case <-discEvent:
+		case ev := <-watcher.Feed():
+			if ev.Type == "add" {
+				activePorts[ev.Port.Address+"|"+ev.Port.Protocol] = ev.Port
+			}
+			if ev.Type == "remove" {
+				delete(activePorts, ev.Port.Address+"|"+ev.Port.Protocol)
+			}
 			updateList()
 		}
 
 		ui.Render(l)
 	}
-
-	for _, disc := range discoveries {
-		disc.Quit()
-		fmt.Println("Discovery QUITed")
-		for disc.State() == discovery.Alive {
-			time.Sleep(time.Millisecond)
-		}
-		fmt.Println("Discovery correctly terminated")
-	}
-
 }

From 7c6784ef5b268cef2cce3f3cc2eae8a80425ce4e Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@arduino.cc>
Date: Mon, 8 Aug 2022 23:32:41 +0200
Subject: [PATCH 14/15] Report discovery start errors

---
 .../discoverymanager/discoverymanager.go          | 15 ++++++++++++---
 cli/arguments/completion.go                       |  2 +-
 cli/arguments/port.go                             |  2 +-
 cli/board/list.go                                 |  5 ++++-
 commands/board/list.go                            | 10 +++++-----
 commands/daemon/daemon.go                         |  2 +-
 6 files changed, 24 insertions(+), 12 deletions(-)

diff --git a/arduino/discovery/discoverymanager/discoverymanager.go b/arduino/discovery/discoverymanager/discoverymanager.go
index 40928544948..5cf3f2d3aa1 100644
--- a/arduino/discovery/discoverymanager/discoverymanager.go
+++ b/arduino/discovery/discoverymanager/discoverymanager.go
@@ -81,11 +81,11 @@ func (dm *DiscoveryManager) IDs() []string {
 
 // Start starts all the discoveries in this DiscoveryManager.
 // If the discoveries are already running, this function does nothing.
-func (dm *DiscoveryManager) Start() {
+func (dm *DiscoveryManager) Start() []error {
 	dm.discoveriesMutex.Lock()
 	defer dm.discoveriesMutex.Unlock()
 	if dm.discoveriesRunning {
-		return
+		return nil
 	}
 
 	go func() {
@@ -95,16 +95,25 @@ func (dm *DiscoveryManager) Start() {
 		}
 	}()
 
+	errs := []error{}
+	var errsLock sync.Mutex
+
 	var wg sync.WaitGroup
 	for _, d := range dm.discoveries {
 		wg.Add(1)
 		go func(d *discovery.PluggableDiscovery) {
-			dm.startDiscovery(d)
+			if err := dm.startDiscovery(d); err != nil {
+				errsLock.Lock()
+				errs = append(errs, err)
+				errsLock.Unlock()
+			}
 			wg.Done()
 		}(d)
 	}
 	wg.Wait()
 	dm.discoveriesRunning = true
+
+	return errs
 }
 
 // Add adds a discovery to the list of managed discoveries
diff --git a/cli/arguments/completion.go b/cli/arguments/completion.go
index 34674e8e57b..32ab99f407b 100644
--- a/cli/arguments/completion.go
+++ b/cli/arguments/completion.go
@@ -178,7 +178,7 @@ func GetInstallableLibs() []string {
 func GetConnectedBoards() []string {
 	inst := instance.CreateAndInit()
 
-	list, _ := board.List(&rpc.BoardListRequest{
+	list, _, _ := board.List(&rpc.BoardListRequest{
 		Instance: inst,
 	})
 	var res []string
diff --git a/cli/arguments/port.go b/cli/arguments/port.go
index d77215dcf58..1e054e713b4 100644
--- a/cli/arguments/port.go
+++ b/cli/arguments/port.go
@@ -146,7 +146,7 @@ func (p *Port) GetSearchTimeout() time.Duration {
 // discovered Port object together with the FQBN. If the port does not match
 // exactly 1 board,
 func (p *Port) DetectFQBN(inst *rpc.Instance) (string, *rpc.Port) {
-	detectedPorts, err := board.List(&rpc.BoardListRequest{
+	detectedPorts, _, err := board.List(&rpc.BoardListRequest{
 		Instance: inst,
 		Timeout:  p.timeout.Get().Milliseconds(),
 	})
diff --git a/cli/board/list.go b/cli/board/list.go
index b162e2a6f79..fe82973847a 100644
--- a/cli/board/list.go
+++ b/cli/board/list.go
@@ -64,13 +64,16 @@ func runListCommand(cmd *cobra.Command, args []string) {
 		os.Exit(0)
 	}
 
-	ports, err := board.List(&rpc.BoardListRequest{
+	ports, discvoeryErrors, err := board.List(&rpc.BoardListRequest{
 		Instance: inst,
 		Timeout:  timeoutArg.Get().Milliseconds(),
 	})
 	if err != nil {
 		feedback.Errorf(tr("Error detecting boards: %v"), err)
 	}
+	for _, err := range discvoeryErrors {
+		feedback.Errorf(tr("Error starting discovery: %v"), err)
+	}
 	feedback.PrintResult(result{ports})
 }
 
diff --git a/commands/board/list.go b/commands/board/list.go
index 84e7149b88f..fe9b4afeb09 100644
--- a/commands/board/list.go
+++ b/commands/board/list.go
@@ -177,21 +177,21 @@ func identify(pm *packagemanager.PackageManager, port *discovery.Port) ([]*rpc.B
 // List returns a list of boards found by the loaded discoveries.
 // In case of errors partial results from discoveries that didn't fail
 // are returned.
-func List(req *rpc.BoardListRequest) (r []*rpc.DetectedPort, e error) {
+func List(req *rpc.BoardListRequest) (r []*rpc.DetectedPort, discoveryStartErrors []error, e error) {
 	pm := commands.GetPackageManager(req.GetInstance().Id)
 	if pm == nil {
-		return nil, &arduino.InvalidInstanceError{}
+		return nil, nil, &arduino.InvalidInstanceError{}
 	}
 
 	dm := pm.DiscoveryManager()
-	dm.Start()
+	discoveryStartErrors = dm.Start()
 	time.Sleep(time.Duration(req.GetTimeout()) * time.Millisecond)
 
 	retVal := []*rpc.DetectedPort{}
 	for _, port := range dm.List() {
 		boards, err := identify(pm, port)
 		if err != nil {
-			return nil, err
+			return nil, discoveryStartErrors, err
 		}
 
 		// boards slice can be empty at this point if neither the cores nor the
@@ -202,7 +202,7 @@ func List(req *rpc.BoardListRequest) (r []*rpc.DetectedPort, e error) {
 		}
 		retVal = append(retVal, b)
 	}
-	return retVal, nil
+	return retVal, discoveryStartErrors, nil
 }
 
 // Watch returns a channel that receives boards connection and disconnection events.
diff --git a/commands/daemon/daemon.go b/commands/daemon/daemon.go
index 03045c87a78..96ec5daed6e 100644
--- a/commands/daemon/daemon.go
+++ b/commands/daemon/daemon.go
@@ -67,7 +67,7 @@ func (s *ArduinoCoreServerImpl) BoardDetails(ctx context.Context, req *rpc.Board
 
 // BoardList FIXMEDOC
 func (s *ArduinoCoreServerImpl) BoardList(ctx context.Context, req *rpc.BoardListRequest) (*rpc.BoardListResponse, error) {
-	ports, err := board.List(req)
+	ports, _, err := board.List(req)
 	if err != nil {
 		return nil, convertErrorToRPCStatus(err)
 	}

From 4c197b8016dad5e8d12f8ed6b01cc02d601b90dd Mon Sep 17 00:00:00 2001
From: Cristian Maglie <c.maglie@bug.st>
Date: Tue, 9 Aug 2022 11:58:43 +0200
Subject: [PATCH 15/15] Update arduino/discovery/discovery_client/main.go

Co-authored-by: Umberto Baldi <34278123+umbynos@users.noreply.github.com>
---
 arduino/discovery/discovery_client/main.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/arduino/discovery/discovery_client/main.go b/arduino/discovery/discovery_client/main.go
index 8306da7462e..b77fe6e7f68 100644
--- a/arduino/discovery/discovery_client/main.go
+++ b/arduino/discovery/discovery_client/main.go
@@ -41,7 +41,7 @@ func main() {
 	activePorts := map[string]*discovery.Port{}
 	watcher, err := dm.Watch()
 	if err != nil {
-		log.Fatalf("failed to start discvoeries: %v", err)
+		log.Fatalf("failed to start discoveries: %v", err)
 	}
 	if err := ui.Init(); err != nil {
 		log.Fatalf("failed to initialize termui: %v", err)