Skip to content

Commit ad47218

Browse files
silvanocerzasilvanocerza
committed
Enhanced handling of failing pluggable discoveries
1 parent 3d5b480 commit ad47218

File tree

3 files changed

+157
-67
lines changed

3 files changed

+157
-67
lines changed

‎arduino/discovery/discoverymanager/discoverymanager.go

Lines changed: 124 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,10 @@
1616
package discoverymanager
1717

1818
import (
19+
"context"
20+
"fmt"
21+
"sync"
22+
1923
"github.com/arduino/arduino-cli/arduino/discovery"
2024
"github.com/pkg/errors"
2125
)
@@ -64,117 +68,187 @@ func (dm *DiscoveryManager) Add(disc *discovery.PluggableDiscovery) error {
6468
return nil
6569
}
6670

67-
// RunAll the discoveries for this DiscoveryManager,
68-
// returns the first error it meets or nil
69-
func (dm *DiscoveryManager) RunAll() error {
71+
// parallelize runs function f concurrently for each discovery.
72+
// Returns a list of errors returned by each call of f.
73+
func (dm *DiscoveryManager) parallelize(f func(d *discovery.PluggableDiscovery) error) []error {
74+
var wg sync.WaitGroup
75+
ctx, cancel := context.WithCancel(context.Background())
76+
errChan := make(chan error)
7077
for _, d := range dm.discoveries {
78+
wg.Add(1)
79+
go func(d *discovery.PluggableDiscovery) {
80+
defer wg.Done()
81+
if err := f(d); err != nil {
82+
errChan <- err
83+
}
84+
}(d)
85+
}
86+
87+
// Wait in a goroutine to collect eventual errors running a discovery.
88+
// When all goroutines that are calling discoveries are done call cancel
89+
// to avoid blocking if there are no errors.
90+
go func() {
91+
wg.Wait()
92+
cancel()
93+
}()
94+
95+
errs := []error{}
96+
for {
97+
select {
98+
case <-ctx.Done():
99+
goto done
100+
case err := <-errChan:
101+
errs = append(errs, err)
102+
}
103+
}
104+
done:
105+
return errs
106+
}
107+
108+
// RunAll the discoveries for this DiscoveryManager,
109+
// returns an error for each discovery failing to run
110+
func (dm *DiscoveryManager) RunAll() []error {
111+
return dm.parallelize(func(d *discovery.PluggableDiscovery) error {
71112
if d.State() != discovery.Dead {
72113
// This discovery is already alive, nothing to do
73-
continue
114+
return nil
74115
}
75116

76117
if err := d.Run(); err != nil {
77-
return err
118+
return fmt.Errorf("discovery %s process not started: %w", d.GetID(), err)
78119
}
79-
}
80-
return nil
120+
return nil
121+
})
81122
}
82123

83124
// StartAll the discoveries for this DiscoveryManager,
84-
// returns the first error it meets or nil
85-
func (dm *DiscoveryManager) StartAll() error {
86-
for _, d := range dm.discoveries {
125+
// returns an error for each discovery failing to start
126+
func (dm *DiscoveryManager) StartAll() []error {
127+
return dm.parallelize(func(d *discovery.PluggableDiscovery) error {
87128
state := d.State()
88129
if state != discovery.Idling || state == discovery.Running {
89130
// Already started
90-
continue
131+
return nil
91132
}
92133
if err := d.Start(); err != nil {
93-
return err
134+
return fmt.Errorf("starting discovery %s: %w", d.GetID(), err)
94135
}
95-
}
96-
return nil
136+
return nil
137+
})
97138
}
98139

99140
// StartSyncAll the discoveries for this DiscoveryManager,
100-
// returns the first error it meets or nil
141+
// returns an error for each discovery failing to start syncing
101142
func (dm *DiscoveryManager) StartSyncAll() (<-chan *discovery.Event, []error) {
102-
errs := []error{}
103143
if dm.globalEventCh == nil {
104144
dm.globalEventCh = make(chan *discovery.Event, 5)
105145
}
106-
for _, d := range dm.discoveries {
146+
errs := dm.parallelize(func(d *discovery.PluggableDiscovery) error {
107147
state := d.State()
108148
if state != discovery.Idling || state == discovery.Syncing {
109149
// Already syncing
110-
continue
150+
return nil
111151
}
112152

113153
eventCh := d.EventChannel(5)
114154
if err := d.StartSync(); err != nil {
115-
errs = append(errs, err)
155+
return fmt.Errorf("start syncing discovery %s: %w", d.GetID(), err)
116156
}
117157
go func() {
118158
for ev := range eventCh {
119159
dm.globalEventCh <- ev
120160
}
121161
}()
122-
}
162+
return nil
163+
})
123164
return dm.globalEventCh, errs
124165
}
125166

126167
// StopAll the discoveries for this DiscoveryManager,
127-
// returns the first error it meets or nil
128-
func (dm *DiscoveryManager) StopAll() error {
129-
for _, d := range dm.discoveries {
168+
// returns an error for each discovery failing to stop
169+
func (dm *DiscoveryManager) StopAll() []error {
170+
return dm.parallelize(func(d *discovery.PluggableDiscovery) error {
130171
state := d.State()
131172
if state != discovery.Syncing && state != discovery.Running {
132173
// Not running nor syncing, nothing to stop
133-
continue
174+
return nil
134175
}
135-
err := d.Stop()
136-
if err != nil {
137-
return err
176+
177+
if err := d.Stop(); err != nil {
178+
return fmt.Errorf("stopping discovery %s: %w", d.GetID(), err)
138179
}
139-
}
140-
return nil
180+
return nil
181+
})
141182
}
142183

143184
// QuitAll quits all the discoveries managed by this DiscoveryManager.
144-
// Returns the first error it meets or nil
145-
func (dm *DiscoveryManager) QuitAll() error {
146-
for _, d := range dm.discoveries {
185+
// Returns an error for each discovery that fails quitting
186+
func (dm *DiscoveryManager) QuitAll() []error {
187+
errs := dm.parallelize(func(d *discovery.PluggableDiscovery) error {
147188
if d.State() == discovery.Dead {
148189
// Stop! Stop! It's already dead!
149-
continue
190+
return nil
150191
}
151-
err := d.Quit()
152-
if err != nil {
153-
return err
192+
193+
if err := d.Quit(); err != nil {
194+
return fmt.Errorf("quitting discovery %s: %w", d.GetID(), err)
154195
}
155-
}
156-
if dm.globalEventCh != nil {
196+
return nil
197+
})
198+
// Close the global channel only if there were no errors
199+
// quitting all alive discoveries
200+
if len(errs) == 0 && dm.globalEventCh != nil {
157201
close(dm.globalEventCh)
158202
dm.globalEventCh = nil
159203
}
160-
return nil
204+
return errs
161205
}
162206

163207
// List returns a list of available ports detected from all discoveries
164-
func (dm *DiscoveryManager) List() []*discovery.Port {
165-
res := []*discovery.Port{}
208+
// and a list of errors for those discoveries that returned one.
209+
func (dm *DiscoveryManager) List() ([]*discovery.Port, []error) {
210+
var wg sync.WaitGroup
211+
// Use this struct to avoid the need of two separate
212+
// channels for ports and errors.
213+
type listMsg struct {
214+
Err error
215+
Port *discovery.Port
216+
}
217+
msgChan := make(chan listMsg)
166218
for _, d := range dm.discoveries {
167-
if d.State() != discovery.Running {
168-
// Discovery is not running, it won't return anything
169-
continue
170-
}
171-
l, err := d.List()
172-
if err != nil {
173-
continue
219+
wg.Add(1)
220+
go func(d *discovery.PluggableDiscovery) {
221+
defer wg.Done()
222+
if d.State() != discovery.Running {
223+
// Discovery is not running, it won't return anything
224+
return
225+
}
226+
ports, err := d.List()
227+
if err != nil {
228+
msgChan <- listMsg{Err: fmt.Errorf("listing ports from discovery %s: %w", d.GetID(), err)}
229+
}
230+
for _, p := range ports {
231+
msgChan <- listMsg{Port: p}
232+
}
233+
}(d)
234+
}
235+
236+
go func() {
237+
// Close the channel only after all goroutines are done
238+
wg.Wait()
239+
close(msgChan)
240+
}()
241+
242+
ports := []*discovery.Port{}
243+
errs := []error{}
244+
for msg := range msgChan {
245+
if msg.Err != nil {
246+
errs = append(errs, msg.Err)
247+
} else {
248+
ports = append(ports, msg.Port)
174249
}
175-
res = append(res, l...)
176250
}
177-
return res
251+
return ports, errs
178252
}
179253

180254
// ListSync return the current list of ports detected from all discoveries

‎cli/arguments/port.go

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222

2323
"github.com/arduino/arduino-cli/arduino/discovery"
2424
"github.com/arduino/arduino-cli/arduino/sketch"
25+
"github.com/arduino/arduino-cli/cli/feedback"
2526
"github.com/arduino/arduino-cli/commands"
2627
rpc "github.com/arduino/arduino-cli/rpc/cc/arduino/cli/commands/v1"
2728
"github.com/pkg/errors"
@@ -76,20 +77,25 @@ func (p *Port) GetPort(instance *rpc.Instance, sk *sketch.Sketch) (*discovery.Po
7677
if pm == nil {
7778
return nil, errors.New("invalid instance")
7879
}
79-
80-
if err := pm.DiscoveryManager().RunAll(); err != nil {
81-
return nil, err
80+
dm := pm.DiscoveryManager()
81+
if errs := dm.RunAll(); len(errs) == len(dm.IDs()) {
82+
// All discoveries failed to run, we can't do anything
83+
return nil, fmt.Errorf("%v", errs)
84+
} else if len(errs) > 0 {
85+
// If only some discoveries failed to run just tell the user and go on
86+
for _, err := range errs {
87+
feedback.Error(err)
88+
}
8289
}
83-
eventChan, errs := pm.DiscoveryManager().StartSyncAll()
90+
eventChan, errs := dm.StartSyncAll()
8491
if len(errs) > 0 {
8592
return nil, fmt.Errorf("%v", errs)
8693
}
8794

8895
defer func() {
8996
// Quit all discoveries at the end.
90-
err := pm.DiscoveryManager().QuitAll()
91-
if err != nil {
92-
logrus.Errorf("quitting discoveries when getting port metadata: %s", err)
97+
if errs := dm.QuitAll(); len(errs) > 0 {
98+
logrus.Errorf("quitting discoveries when getting port metadata: %v", errs)
9399
}
94100
}()
95101

‎commands/board/list.go

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -188,21 +188,25 @@ func List(req *rpc.BoardListRequest) (r []*rpc.DetectedPort, e error) {
188188
}
189189

190190
dm := pm.DiscoveryManager()
191-
if err := dm.RunAll(); err != nil {
192-
return nil, err
191+
if errs := dm.RunAll(); len(errs) > 0 {
192+
return nil, fmt.Errorf("%v", errs)
193193
}
194-
if err := dm.StartAll(); err != nil {
195-
return nil, err
194+
if errs := dm.StartAll(); len(errs) > 0 {
195+
return nil, fmt.Errorf("%v", errs)
196196
}
197197
defer func() {
198-
if err := dm.StopAll(); err != nil {
199-
logrus.Error(err)
198+
if errs := dm.StopAll(); len(errs) > 0 {
199+
logrus.Error(errs)
200200
}
201201
}()
202202
time.Sleep(time.Duration(req.GetTimeout()) * time.Millisecond)
203203

204204
retVal := []*rpc.DetectedPort{}
205-
for _, port := range pm.DiscoveryManager().List() {
205+
ports, errs := pm.DiscoveryManager().List()
206+
if len(errs) > 0 {
207+
return nil, fmt.Errorf("%v", errs)
208+
}
209+
for _, port := range ports {
206210
boards, err := identify(pm, port)
207211
if err != nil {
208212
return nil, err
@@ -224,12 +228,18 @@ func List(req *rpc.BoardListRequest) (r []*rpc.DetectedPort, e error) {
224228
// The discovery process can be interrupted by sending a message to the interrupt channel.
225229
func Watch(instanceID int32, interrupt <-chan bool) (<-chan *rpc.BoardListWatchResponse, error) {
226230
pm := commands.GetPackageManager(instanceID)
231+
dm := pm.DiscoveryManager()
227232

228-
if err := pm.DiscoveryManager().RunAll(); err != nil {
229-
return nil, err
233+
runErrs := dm.RunAll()
234+
if len(runErrs) == len(dm.IDs()) {
235+
// All discoveries failed to run, we can't do anything
236+
return nil, fmt.Errorf("%v", runErrs)
230237
}
231238

232-
eventsChan, errs := pm.DiscoveryManager().StartSyncAll()
239+
eventsChan, errs := dm.StartSyncAll()
240+
if len(runErrs) > 0 {
241+
errs = append(runErrs, errs...)
242+
}
233243

234244
outChan := make(chan *rpc.BoardListWatchResponse)
235245

0 commit comments

Comments
 (0)