Commit 35465c8e7664e9ced56c32eb410462446974baa7 - golang-github-vbauerster-mpb

+5

-3

bar.go less more

1	1
2	2	import (
3	3	"bytes"
	4	"context"
4	5	"fmt"
5	6	"io"
6	7	"io/ioutil"

75	76	)
76	77
77	78	func newBar(
	79	ctx context.Context,
78	80	wg *sync.WaitGroup,
79	81	filler Filler,
80	82	id, width int,
81	83	total int64,
82		cancel <-chan struct{},
83	84	options ...BarOption,
84	85	) *Bar {
85	86	if total <= 0 {

123	124	b.priority = b.runningBar.priority
124	125	}
125	126
126		go b.serve(wg, s, cancel)
	127	go b.serve(ctx, wg, s)
127	128	return b
128	129	}
129	130

248	249	}
249	250	}
250	251
251		func (b Bar) serve(wg sync.WaitGroup, s *bState, cancel <-chan struct{}) {
	252	func (b Bar) serve(ctx context.Context, wg sync.WaitGroup, s *bState) {
252	253	defer wg.Done()
	254	cancel := ctx.Done()
253	255	for {
254	256	select {
255	257	case op := <-b.operateState:

+0

-2

examples/cancel/main.go less more

+11

-8

options.go less more

0	0	package mpb
1	1
2	2	import (
	3	"context"
3	4	"io"
4	5	"sync"
5	6	"time"

8	9	)
9	10
10	11	// ProgressOption is a function option which changes the default behavior of
11		// progress pool, if passed to mpb.New(...ProgressOption)
	12	// progress pool, if passed to mpb.New(...ProgressOption).
12	13	type ProgressOption func(*pState)
13	14
14	15	// WithWaitGroup provides means to have a single joint point.

21	22	}
22	23	}
23	24
24		// WithWidth overrides default width 80
	25	// WithWidth overrides default width 80.
25	26	func WithWidth(w int) ProgressOption {
26	27	return func(s *pState) {
27	28	if w >= 0 {

30	31	}
31	32	}
32	33
33		// WithRefreshRate overrides default 120ms refresh rate
	34	// WithRefreshRate overrides default 120ms refresh rate.
34	35	func WithRefreshRate(d time.Duration) ProgressOption {
35	36	return func(s *pState) {
36	37	if d < 10*time.Millisecond {

48	49	}
49	50	}
50	51
51		// WithCancel provide your cancel channel,
52		// which you plan to close at some point.
53		func WithCancel(ch <-chan struct{}) ProgressOption {
	52	// WithContext provided context will be used for cancellation purposes.
	53	func WithContext(ctx context.Context) ProgressOption {
54	54	return func(s *pState) {
55		s.cancel = ch
	55	if ctx == nil {
	56	return
	57	}
	58	s.ctx = ctx
56	59	}
57	60	}
58	61

63	66	}
64	67	}
65	68
66		// WithOutput overrides default output os.Stdout
	69	// WithOutput overrides default output os.Stdout.
67	70	func WithOutput(w io.Writer) ProgressOption {
68	71	return func(s *pState) {
69	72	if w == nil {

+0

-15

~~options_go1.7.go~~ less more

0		//+build go1.7
1
2		package mpb
3
4		import "context"
5
6		// WithContext provided context will be used for cancellation purposes
7		func WithContext(ctx context.Context) ProgressOption {
8		return func(s *pState) {
9		if ctx == nil {
10		panic("ctx must not be nil")
11		}
12		s.cancel = ctx.Done()
13		}
14		}

+5

-3

progress.go less more

1	1
2	2	import (
3	3	"container/heap"
	4	"context"
4	5	"fmt"
5	6	"io"
6	7	"io/ioutil"

39	40	pMatrix map[int][]chan int
40	41	aMatrix map[int][]chan int
41	42
42		// following are provided by user
	43	// following are provided/overrided by user
	44	ctx context.Context
43	45	uwg *sync.WaitGroup
44	46	manualRefreshCh <-chan time.Time
45		cancel <-chan struct{}
46	47	shutdownNotifier chan struct{}
47	48	waitBars map[Bar]Bar
48	49	debugOut io.Writer

54	55	pq := make(priorityQueue, 0)
55	56	heap.Init(&pq)
56	57	s := &pState{
	58	ctx: context.Background(),
57	59	bHeap: &pq,
58	60	width: pwidth,
59	61	cw: cwriter.New(os.Stdout),

100	102	result := make(chan *Bar)
101	103	select {
102	104	case p.operateState <- func(s *pState) {
103		b := newBar(p.wg, filler, s.idCounter, s.width, total, s.cancel, options...)
	105	b := newBar(s.ctx, p.wg, filler, s.idCounter, s.width, total, options...)
104	106	if b.runningBar != nil {
105	107	s.waitBars[b.runningBar] = b
106	108	} else {

+0

-50

~~progress_go1.7_test.go~~ less more

0		//+build go1.7
1
2		package mpb_test
3
4		import (
5		"context"
6		"io/ioutil"
7		"testing"
8		"time"
9
10		"github.com/vbauerster/mpb"
11		)
12
13		func TestWithContext(t *testing.T) {
14		ctx, cancel := context.WithCancel(context.Background())
15		shutdown := make(chan struct{})
16		p := mpb.New(
17		mpb.WithOutput(ioutil.Discard),
18		mpb.WithContext(ctx),
19		mpb.WithShutdownNotifier(shutdown),
20		)
21
22		total := 1000
23		numBars := 3
24		bars := make([]*mpb.Bar, 0, numBars)
25		for i := 0; i < numBars; i++ {
26		bar := p.AddBar(int64(total))
27		bars = append(bars, bar)
28		go func() {
29		for !bar.Completed() {
30		time.Sleep(randomDuration(40 * time.Millisecond))
31		bar.Increment()
32		}
33		}()
34		}
35
36		time.AfterFunc(100*time.Millisecond, cancel)
37
38		p.Wait()
39		for _, bar := range bars {
40		if bar.Current() >= int64(total) {
41		t.Errorf("bar %d: total = %d, current = %d\n", bar.ID(), total, bar.Current())
42		}
43		}
44		select {
45		case <-shutdown:
46		case <-time.After(100 * time.Millisecond):
47		t.Error("Progress didn't stop")
48		}
49		}

+20

-15

progress_test.go less more

1	1
2	2	import (
3	3	"bytes"
	4	"context"
4	5	"fmt"
5	6	"io/ioutil"
6	7	"math/rand"

8	9	"testing"
9	10	"time"
10	11
	12	"github.com/vbauerster/mpb"
11	13	. "github.com/vbauerster/mpb"
12	14	"github.com/vbauerster/mpb/cwriter"
13	15	)

79	81	p.Wait()
80	82	}
81	83
82		func TestWithCancel(t *testing.T) {
83		cancel := make(chan struct{})
	84	func TestWithContext(t *testing.T) {
	85	ctx, cancel := context.WithCancel(context.Background())
84	86	shutdown := make(chan struct{})
85		p := New(
86		WithOutput(ioutil.Discard),
87		WithCancel(cancel),
88		WithShutdownNotifier(shutdown),
	87	p := mpb.New(
	88	mpb.WithOutput(ioutil.Discard),
	89	mpb.WithContext(ctx),
	90	mpb.WithRefreshRate(50*time.Millisecond),
	91	mpb.WithShutdownNotifier(shutdown),
89	92	)
90	93
91		for i := 0; i < 2; i++ {
92		bar := p.AddBar(int64(1000), BarID(i))
	94	total := 10000
	95	numBars := 3
	96	bars := make([]*mpb.Bar, 0, numBars)
	97	for i := 0; i < numBars; i++ {
	98	bar := p.AddBar(int64(total))
	99	bars = append(bars, bar)
93	100	go func() {
94	101	for !bar.Completed() {
	102	bar.Increment()
95	103	time.Sleep(randomDuration(100 * time.Millisecond))
96		bar.Increment()
97	104	}
98	105	}()
99	106	}
100	107
101		time.AfterFunc(100*time.Millisecond, func() {
102		close(cancel)
103		})
	108	time.Sleep(50 * time.Millisecond)
	109	cancel()
104	110
105	111	p.Wait()
106
107	112	select {
108	113	case <-shutdown:
109		case <-time.After(200 * time.Millisecond):
110		t.FailNow()
	114	case <-time.After(100 * time.Millisecond):
	115	t.Error("Progress didn't stop")
111	116	}
112	117	}
113	118