1266

Committed 05 Apr 2023 11:52PM UTC coverage: 94.605% (+0.03%) from 94.572%

Build # 1266

Build Type

push

travis-ci-com

Committed by GitHub

Commit Message

Do not log raw table metadata (#168)

* Do not log raw table metadata

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

2227 of 2354 relevant lines covered (94.6%)

155.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.75

/memoryless/memoryless.go

// Package memoryless helps repeated calls to a function be distributed across
// time in a memoryless fashion.
package memoryless

import (
        "context"
        "fmt"
        "math/rand"
        "time"
)

// Config represents the time we should wait between runs of the function.
//
// A valid config will have:
//  0 <= Min <= Expected <= Max (or 0 <= Min <= Expected and Max is 0)
// If Max is zero or unset, it will be ignored. If Min is zero or unset, it will
// be ignored.
type Config struct {
        // Expected records the expected/mean/average amount of time between runs.
        Expected time.Duration
        // Min provides clamping of the randomly produced value. All timers will wait
        // at least Min time.
        Min time.Duration
        // Max provides clamping of the randomly produced value. All timers will take
        // at most Max time.
        Max time.Duration

        // Once is provided as a helper, because frequently for unit testing and
        // integration testing, you only want the "Forever" loop to run once.
        //
        // The zero value of this struct has Once set to false, which means the value
        // only needs to be set explicitly in codepaths where it might be true.
        Once bool
}

func (c Config) waittime() time.Duration {
        wt := time.Duration(rand.ExpFloat64() * float64(c.Expected))
        if wt < c.Min {
                wt = c.Min
        }
        if c.Max != 0 && wt > c.Max {
                wt = c.Max
        }
        return wt
}

// Check whether the config contrains sensible values. It return an error if the
// config makes no mathematical sense, and nil if everything is okay.
func (c Config) Check() error {
        if !(0 <= c.Min && c.Min <= c.Expected && (c.Max == 0 || c.Expected <= c.Max)) {
                return fmt.Errorf(
                        "The arguments to Run make no sense. It should be true that Min <= Expected <= Max (or Min <= Expected and Max is 0), "+
                                "but that is not true for Min(%v) Expected(%v) Max(%v).",
                        c.Min, c.Expected, c.Max)
        }
        return nil
}

// newTimer constructs and returns a timer. This function assumes that the
// config has no errors.
func newTimer(c Config) *time.Timer {
        return time.NewTimer(c.waittime())
}

// NewTimer constructs a single-shot time.Timer that, if repeatedly used to
// construct a series of timers, will ensure that the resulting events conform
// to the memoryless distribution. For more on how this could and should be
// used, see the comments to Ticker. It is intended to be a drop-in replacement
// for time.NewTimer.
func NewTimer(c Config) (*time.Timer, error) {
        if err := c.Check(); err != nil {
                return nil, err
        }

        return newTimer(c), nil
}

// AfterFunc constructs a single-shot time.Timer that, if repeatedly used to
// construct a series of timers, will ensure that the resulting events conform
// to the memoryless distribution. For more on how this could and should be
// used, see the comments to Ticker. It is intended to be a drop-in replacement
// for time.AfterFunc.
func AfterFunc(c Config, f func()) (*time.Timer, error) {
        if err := c.Check(); err != nil {
                return nil, err
        }

        return time.AfterFunc(c.waittime(), f), nil
}

// Ticker is a struct that waits a config.Expected amount of time on average
// between sends down the channel C. It has the same interface and requirements
// as time.Ticker. Every Ticker created must have its Stop() method called or it
// will leak a goroutine.
//
// The inter-send time is a random variable governed by the exponential
// distribution and will generate a memoryless (Poisson) distribution of channel
// reads over time, ensuring that a measurement scheme using this ticker has the
// PASTA property (Poisson Arrivals See Time Averages). This statistical
// guarantee is subject to two caveats:
//
// Caveat 1 is that, in a nod to the realities of systems needing to have
// guarantees, we allow the random wait time to be clamped both above and below.
// This means that channel events will be at least config.Min and at most
// config.Max apart in time. This clamping causes bias in the timing. For use of
// Ticker to be statistically sensible, the clamping should not be too extreme.
// The exact mathematical meaning of "too extreme" depends on your situation,
// but a nice rule of thumb is config.Min should be at most 10% of expected and
// config.Max should be at least 250% of expected. These values mean that less
// than 10% of time you will be waiting config.Min and less than 10% of the time
// you will be waiting config.Max.
//
// Caveat 2 is that this assumes that the actions performed between channel
// reads take negligible time when compared to the expected wait time.
// Memoryless sequences have the property that the times between successive
// event starts has the exponential distribution, and the exponential
// distribution can generate numbers arbitrarily close to zero (albeit
// exponentially infrequently). This code will not send on the channel if the
// other end is not ready to receive, which provides another lower bound on
// inter-event times. The only other option if the other side of the channel is
// not ready to receive would be queueing events in the channel, and that has
// some pathological cases we would like to avoid. In particular, queuing can
// cause long-term correlations if the queue gets big, which is the exact
// opposite of what a memoryless system is trying to achieve.
type Ticker struct {
        C         <-chan time.Time // The channel on which the ticks are delivered.
        config    Config
        writeChan chan<- time.Time
        cancel    func()
}

func (t *Ticker) singleIteration(ctx context.Context) {
        timer := newTimer(t.config)
        defer timer.Stop()
        // Wait until the timer is done or the context is canceled. If both conditions
        // are true, which case gets called is unspecified.
        select {
        case <-ctx.Done():
                // Please don't put code here that assumes that this code path will
                // definitely execute if the context is done. select {} doesn't promise that
                // multiple channels will get selected with equal probability, which means
                // that it could be true that the timer is done AND the context is canceled,
                // and we have no guarantee that in that case the canceled context case will
                // be the one that is selected.
        case <-timer.C:
        }
        // Just like time.Ticker, writes to the channel are non-blocking. If a user of
        // this module can't keep up with the timer they set, that's on them. There
        // are some potential pathological cases associated with queueing events in
        // the channel, and we want to avoid them.
        select {
        case t.writeChan <- time.Now():
        default:
        }
}

func (t *Ticker) runTicker(ctx context.Context) {
        // No matter what, when this function exits the channel should never be written to again.
        defer close(t.writeChan)

        if t.config.Once {
                if ctx.Err() == nil {
                        t.singleIteration(ctx)
                }
                return
        }

        // When Done() is not closed and the Deadline has not been exceeded, the error
        // is nil.
        for ctx.Err() == nil {
                t.singleIteration(ctx)
        }
}

// Stop the ticker goroutine.
func (t *Ticker) Stop() {
        t.cancel()
}

// MakeTicker is a deprecated alias for NewTicker
var MakeTicker = NewTicker

// NewTicker creates a new memoryless ticker. The returned struct is compatible
// with the time.Ticker struct interface, and everywhere you use a time.Ticker,
// you can use a memoryless.Ticker.
func NewTicker(ctx context.Context, config Config) (*Ticker, error) {
        if err := config.Check(); err != nil {
                return nil, err
        }
        c := make(chan time.Time)
        ctx, cancel := context.WithCancel(ctx)
        ticker := &Ticker{
                C:         c,
                config:    config,
                writeChan: c,
                cancel:    cancel,
        }
        go ticker.runTicker(ctx)
        return ticker, nil
}

// Run calls the given function repeatedly, using a memoryless.Ticker to wait
// between function calls. It is a convenience function for code that does not
// want to use the channel interface.
func Run(ctx context.Context, f func(), c Config) error {
        ticker, err := MakeTicker(ctx, c)
        if err != nil {
                return err
        }
        defer ticker.Stop()
        for range ticker.C {
                f()
        }
        return nil
}

1	// Package memoryless helps repeated calls to a function be distributed across
2	// time in a memoryless fashion.
3	package memoryless
4
5	import (
6	"context"
7	"fmt"
8	"math/rand"
9	"time"
10	)
11
12	// Config represents the time we should wait between runs of the function.
13	//
14	// A valid config will have:
15	// 0 <= Min <= Expected <= Max (or 0 <= Min <= Expected and Max is 0)
16	// If Max is zero or unset, it will be ignored. If Min is zero or unset, it will
17	// be ignored.
18	type Config struct {
19	// Expected records the expected/mean/average amount of time between runs.
20	Expected time.Duration
21	// Min provides clamping of the randomly produced value. All timers will wait
22	// at least Min time.
23	Min time.Duration
24	// Max provides clamping of the randomly produced value. All timers will take
25	// at most Max time.
26	Max time.Duration
27
28	// Once is provided as a helper, because frequently for unit testing and
29	// integration testing, you only want the "Forever" loop to run once.
30	//
31	// The zero value of this struct has Once set to false, which means the value
32	// only needs to be set explicitly in codepaths where it might be true.
33	Once bool
34	}
35
36	func (c Config) waittime() time.Duration {	1,188✔
37	wt := time.Duration(rand.ExpFloat64() * float64(c.Expected))	1,188✔
38	if wt < c.Min {	1,958✔
39	wt = c.Min	770✔
40	}	770✔
41	if c.Max != 0 && wt > c.Max {	1,605✔
42	wt = c.Max	417✔
43	}	417✔
44	return wt	1,188✔
45	}
46
47	// Check whether the config contrains sensible values. It return an error if the
48	// config makes no mathematical sense, and nil if everything is okay.
49	func (c Config) Check() error {	45✔
50	if !(0 <= c.Min && c.Min <= c.Expected && (c.Max == 0 \|\| c.Expected <= c.Max)) {	85✔
51	return fmt.Errorf(	40✔
52	"The arguments to Run make no sense. It should be true that Min <= Expected <= Max (or Min <= Expected and Max is 0), "+	40✔
53	"but that is not true for Min(%v) Expected(%v) Max(%v).",	40✔
54	c.Min, c.Expected, c.Max)	40✔
55	}	40✔
56	return nil	5✔
57	}
58
59	// newTimer constructs and returns a timer. This function assumes that the
60	// config has no errors.
61	func newTimer(c Config) *time.Timer {	1,187✔
62	return time.NewTimer(c.waittime())	1,187✔
63	}	1,187✔
64
65	// NewTimer constructs a single-shot time.Timer that, if repeatedly used to
66	// construct a series of timers, will ensure that the resulting events conform
67	// to the memoryless distribution. For more on how this could and should be
68	// used, see the comments to Ticker. It is intended to be a drop-in replacement
69	// for time.NewTimer.
70	func NewTimer(c Config) (*time.Timer, error) {	9✔
71	if err := c.Check(); err != nil {	17✔
72	return nil, err	8✔
73	}	8✔
74
75	return newTimer(c), nil	1✔
76	}
77
78	// AfterFunc constructs a single-shot time.Timer that, if repeatedly used to
79	// construct a series of timers, will ensure that the resulting events conform
80	// to the memoryless distribution. For more on how this could and should be
81	// used, see the comments to Ticker. It is intended to be a drop-in replacement
82	// for time.AfterFunc.
83	func AfterFunc(c Config, f func()) (*time.Timer, error) {	9✔
84	if err := c.Check(); err != nil {	17✔
85	return nil, err	8✔
86	}	8✔
87
88	return time.AfterFunc(c.waittime(), f), nil	1✔
89	}
90
91	// Ticker is a struct that waits a config.Expected amount of time on average
92	// between sends down the channel C. It has the same interface and requirements
93	// as time.Ticker. Every Ticker created must have its Stop() method called or it
94	// will leak a goroutine.
95	//
96	// The inter-send time is a random variable governed by the exponential
97	// distribution and will generate a memoryless (Poisson) distribution of channel
98	// reads over time, ensuring that a measurement scheme using this ticker has the
99	// PASTA property (Poisson Arrivals See Time Averages). This statistical
100	// guarantee is subject to two caveats:
101	//
102	// Caveat 1 is that, in a nod to the realities of systems needing to have
103	// guarantees, we allow the random wait time to be clamped both above and below.
104	// This means that channel events will be at least config.Min and at most
105	// config.Max apart in time. This clamping causes bias in the timing. For use of
106	// Ticker to be statistically sensible, the clamping should not be too extreme.
107	// The exact mathematical meaning of "too extreme" depends on your situation,
108	// but a nice rule of thumb is config.Min should be at most 10% of expected and
109	// config.Max should be at least 250% of expected. These values mean that less
110	// than 10% of time you will be waiting config.Min and less than 10% of the time
111	// you will be waiting config.Max.
112	//
113	// Caveat 2 is that this assumes that the actions performed between channel
114	// reads take negligible time when compared to the expected wait time.
115	// Memoryless sequences have the property that the times between successive
116	// event starts has the exponential distribution, and the exponential
117	// distribution can generate numbers arbitrarily close to zero (albeit
118	// exponentially infrequently). This code will not send on the channel if the
119	// other end is not ready to receive, which provides another lower bound on
120	// inter-event times. The only other option if the other side of the channel is
121	// not ready to receive would be queueing events in the channel, and that has
122	// some pathological cases we would like to avoid. In particular, queuing can
123	// cause long-term correlations if the queue gets big, which is the exact
124	// opposite of what a memoryless system is trying to achieve.
125	type Ticker struct {
126	C <-chan time.Time // The channel on which the ticks are delivered.
127	config Config
128	writeChan chan<- time.Time
129	cancel func()
130	}
131
132	func (t *Ticker) singleIteration(ctx context.Context) {	1,186✔
133	timer := newTimer(t.config)	1,186✔
134	defer timer.Stop()	1,186✔
135	// Wait until the timer is done or the context is canceled. If both conditions	1,186✔
136	// are true, which case gets called is unspecified.	1,186✔
137	select {	1,186✔
138	case <-ctx.Done():	×
139	// Please don't put code here that assumes that this code path will
140	// definitely execute if the context is done. select {} doesn't promise that
141	// multiple channels will get selected with equal probability, which means
142	// that it could be true that the timer is done AND the context is canceled,
143	// and we have no guarantee that in that case the canceled context case will
144	// be the one that is selected.
145	case <-timer.C:	1,186✔
146	}
147	// Just like time.Ticker, writes to the channel are non-blocking. If a user of
148	// this module can't keep up with the timer they set, that's on them. There
149	// are some potential pathological cases associated with queueing events in
150	// the channel, and we want to avoid them.
151	select {	1,186✔
152	case t.writeChan <- time.Now():	1,003✔
153	default:	183✔
154	}
155	}
156
157	func (t *Ticker) runTicker(ctx context.Context) {	3✔
158	// No matter what, when this function exits the channel should never be written to again.	3✔
159	defer close(t.writeChan)	3✔
160		3✔
161	if t.config.Once {	4✔
162	if ctx.Err() == nil {	2✔
163	t.singleIteration(ctx)	1✔
164	}	1✔
165	return	1✔
166	}
167
168	// When Done() is not closed and the Deadline has not been exceeded, the error
169	// is nil.
170	for ctx.Err() == nil {	1,187✔
171	t.singleIteration(ctx)	1,185✔
172	}	1,185✔
173	}
174
175	// Stop the ticker goroutine.
176	func (t *Ticker) Stop() {	3✔
177	t.cancel()	3✔
178	}	3✔
179
180	// MakeTicker is a deprecated alias for NewTicker
181	var MakeTicker = NewTicker
182
183	// NewTicker creates a new memoryless ticker. The returned struct is compatible
184	// with the time.Ticker struct interface, and everywhere you use a time.Ticker,
185	// you can use a memoryless.Ticker.
186	func NewTicker(ctx context.Context, config Config) (*Ticker, error) {	19✔
187	if err := config.Check(); err != nil {	35✔
188	return nil, err	16✔
189	}	16✔
190	c := make(chan time.Time)	3✔
191	ctx, cancel := context.WithCancel(ctx)	3✔
192	ticker := &Ticker{	3✔
193	C: c,	3✔
194	config: config,	3✔
195	writeChan: c,	3✔
196	cancel: cancel,	3✔
197	}	3✔
198	go ticker.runTicker(ctx)	3✔
199	return ticker, nil	3✔
200	}
201
202	// Run calls the given function repeatedly, using a memoryless.Ticker to wait
203	// between function calls. It is a convenience function for code that does not
204	// want to use the channel interface.
205	func Run(ctx context.Context, f func(), c Config) error {	10✔
206	ticker, err := MakeTicker(ctx, c)	10✔
207	if err != nil {	18✔
208	return err	8✔
209	}	8✔
210	defer ticker.Stop()	2✔
211	for range ticker.C {	1,005✔
212	f()	1,003✔
213	}	1,003✔
214	return nil	2✔
215	}

m-lab / go / 1266

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous