Codebase list golang-github-vbauerster-mpb / 208ad3a
wait wor syncWidth before next render Vladimir Bauer 3 years ago
4 changed file(s) with 35 addition(s) and 100 deletion(s). Raw diff Collapse all Expand all
+14
-18
decorators_test.go less more
182182 }
183183
184184 for _, columnCase := range testCases {
185 mpb.SyncWidth(toSyncMatrix(columnCase))
186 numBars := len(columnCase)
187 gott := make([]chan string, numBars)
188 wg := new(sync.WaitGroup)
189 wg.Add(numBars)
190 for i, step := range columnCase {
185 var wg sync.WaitGroup
186 mpb.SyncWidth(&wg, toSyncMatrix(columnCase))
187 var results []chan string
188 for _, step := range columnCase {
191189 step := step
192 ch := make(chan string, 1)
190 ch := make(chan string)
193191 go func() {
194 defer wg.Done()
195192 ch <- step.decorator.Decor(step.stat)
196193 }()
197 gott[i] = ch
194 results = append(results, ch)
195 }
196
197 for i, ch := range results {
198 res := <-ch
199 want := columnCase[i].want
200 if res != want {
201 t.Errorf("Want: %q, Got: %q\n", want, res)
202 }
198203 }
199204 wg.Wait()
200
201 for i, ch := range gott {
202 got := <-ch
203 want := columnCase[i].want
204 if got != want {
205 t.Errorf("Want: %q, Got: %q\n", want, got)
206 }
207 }
208
209205 }
210206 }
211207
+0
-1
export_test.go less more
11
22 // make syncWidth func public in test
33 var SyncWidth = syncWidth
4 var MaxWidthDistributor = maxWidthDistributor
+21
-17
progress.go less more
296296 }
297297
298298 func (s *pState) render(cw *cwriter.Writer) error {
299 var wg sync.WaitGroup
299300 if s.heapUpdated {
300301 s.updateSyncMatrix()
301302 s.heapUpdated = false
302303 }
303 syncWidth(s.pMatrix)
304 syncWidth(s.aMatrix)
304 syncWidth(&wg, s.pMatrix)
305 syncWidth(&wg, s.aMatrix)
305306
306307 width, height, err := cw.GetTermSize()
307308 if err != nil {
313314 go bar.render(width)
314315 }
315316
316 return s.flush(cw, height)
317 }
318
319 func (s *pState) flush(cw *cwriter.Writer, height int) error {
320 var wg sync.WaitGroup
317 err = s.flush(&wg, cw, height)
318 wg.Wait()
319 return err
320 }
321
322 func (s *pState) flush(wg *sync.WaitGroup, cw *cwriter.Writer, height int) error {
321323 var popCount int
322324
323325 for s.bHeap.Len() > 0 {
372374 }
373375 case 1:
374376 heap.Push(&s.bHeap, s.pool[0])
377 s.pool = s.pool[:0]
375378 default:
376379 wg.Add(1)
377380 go func() {
378381 for _, b := range s.pool {
379382 heap.Push(&s.bHeap, b)
380383 }
384 s.pool = s.pool[:0]
381385 wg.Done()
382386 }()
383387 }
385389 for i := len(s.rows) - 1; i >= 0; i-- {
386390 _, err := cw.ReadFrom(s.rows[i])
387391 if err != nil {
388 wg.Wait()
392 s.rows = s.rows[:0]
389393 return err
390394 }
391395 }
392396
393397 err := cw.Flush(len(s.rows) - popCount)
394 wg.Wait()
395398 s.rows = s.rows[:0]
396 s.pool = s.pool[:0]
397399 return err
398400 }
399401
482484 return bs
483485 }
484486
485 func syncWidth(matrix map[int][]chan int) {
487 func syncWidth(wg *sync.WaitGroup, matrix map[int][]chan int) {
486488 for _, column := range matrix {
487 go maxWidthDistributor(column)
488 }
489 }
490
491 func maxWidthDistributor(column []chan int) {
489 wg.Add(1)
490 go maxWidthDistributor(wg, column)
491 }
492 }
493
494 func maxWidthDistributor(wg *sync.WaitGroup, column []chan int) {
492495 var maxWidth int
493496 for _, ch := range column {
494497 if w := <-ch; w > maxWidth {
498501 for _, ch := range column {
499502 ch <- maxWidth
500503 }
501 }
504 wg.Done()
505 }
+0
-64
progress_test.go less more
111111 }
112112 }
113113
114 func TestMaxWidthDistributor(t *testing.T) {
115 makeWrapper := func(f func([]chan int), start, end chan struct{}) func([]chan int) {
116 return func(column []chan int) {
117 start <- struct{}{}
118 f(column)
119 <-end
120 }
121 }
122
123 ready := make(chan struct{})
124 start := make(chan struct{})
125 end := make(chan struct{})
126 // mpb.MaxWidthDistributor shouldn't stuck in the middle while removing or aborting a bar
127 mpb.MaxWidthDistributor = makeWrapper(mpb.MaxWidthDistributor, start, end)
128
129 total := 100
130 numBars := 6
131 p := mpb.New(mpb.WithOutput(io.Discard))
132 for i := 0; i < numBars; i++ {
133 bar := p.AddBar(int64(total),
134 mpb.BarOptional(mpb.BarRemoveOnComplete(), i == 0),
135 mpb.PrependDecorators(decor.EwmaETA(decor.ET_STYLE_GO, 60, decor.WCSyncSpace)),
136 )
137 go func() {
138 <-ready
139 for i := 0; i < total; i++ {
140 start := time.Now()
141 if id := bar.ID(); id > 1 && i >= 32 {
142 if id&1 == 1 {
143 bar.Abort(true)
144 } else {
145 bar.Abort(false)
146 }
147 }
148 time.Sleep(randomDuration(100 * time.Millisecond))
149 bar.EwmaIncrInt64(rand.Int63n(5)+1, time.Since(start))
150 }
151 }()
152 }
153
154 go func() {
155 <-ready
156 p.Wait()
157 close(start)
158 }()
159
160 res := t.Run("maxWidthDistributor", func(t *testing.T) {
161 close(ready)
162 for v := range start {
163 timer := time.NewTimer(100 * time.Millisecond)
164 select {
165 case end <- v:
166 timer.Stop()
167 case <-timer.C:
168 t.FailNow()
169 }
170 }
171 })
172
173 if !res {
174 t.Error("maxWidthDistributor stuck in the middle")
175 }
176 }
177
178114 func TestProgressShutdownsWithErrFiller(t *testing.T) {
179115 var debug bytes.Buffer
180116 shutdown := make(chan struct{})