Line data    Source code

       1             : package errorfs
       2             : 
       3             : import (
       4             :         "encoding/binary"
       5             :         "fmt"
       6             :         "go/token"
       7             :         "hash/maphash"
       8             :         "math/rand"
       9             :         "strconv"
      10             :         "sync"
      11             :         "time"
      12             : 
      13             :         "github.com/cockroachdb/errors"
      14             :         "github.com/cockroachdb/pebble/internal/dsl"
      15             : )
      16             : 
      17             : // RandomLatency constructs an Injector that does not inject errors but instead
      18             : // injects random latency into operations that match the provided predicate. The
      19             : // amount of latency injected follows an exponential distribution with the
      20             : // provided mean. Latency injected is derived from the provided seed and is
      21             : // deterministic with respect to each file's path.
      22           1 : func RandomLatency(pred Predicate, mean time.Duration, seed int64) Injector {
      23           1 :         rl := &randomLatency{
      24           1 :                 predicate: pred,
      25           1 :                 mean:      mean,
      26           1 :         }
      27           1 :         rl.keyedPrng.init(seed)
      28           1 :         return rl
      29           1 : }
      30             : 
      31           0 : func parseRandomLatency(p *Parser, s *dsl.Scanner) Injector {
      32           0 :         dur, err := time.ParseDuration(s.ConsumeString())
      33           0 :         if err != nil {
      34           0 :                 panic(errors.Newf("parsing RandomLatency: %s", err))
      35             :         }
      36           0 :         lit := s.Consume(token.INT).Lit
      37           0 :         seed, err := strconv.ParseInt(lit, 10, 64)
      38           0 :         if err != nil {
      39           0 :                 panic(err)
      40             :         }
      41           0 :         var pred Predicate
      42           0 :         tok := s.Scan()
      43           0 :         if tok.Kind == token.LPAREN || tok.Kind == token.IDENT {
      44           0 :                 pred = p.predicates.ParseFromPos(s, tok)
      45           0 :                 tok = s.Scan()
      46           0 :         }
      47           0 :         if tok.Kind != token.RPAREN {
      48           0 :                 panic(errors.Errorf("errorfs: unexpected token %s; expected %s", tok.String(), token.RPAREN))
      49             :         }
      50           0 :         return RandomLatency(pred, dur, seed)
      51             : }
      52             : 
      53             : type randomLatency struct {
      54             :         predicate Predicate
      55             :         // p defines the probability of an error being injected.
      56             :         mean time.Duration
      57             :         keyedPrng
      58             : }
      59             : 
      60           0 : func (rl *randomLatency) String() string {
      61           0 :         if rl.predicate == nil {
      62           0 :                 return fmt.Sprintf("(RandomLatency %q %d)", rl.mean, rl.rootSeed)
      63           0 :         }
      64           0 :         return fmt.Sprintf("(RandomLatency %q %d %s)", rl.mean, rl.rootSeed, rl.predicate)
      65             : }
      66             : 
      67           1 : func (rl *randomLatency) MaybeError(op Op) error {
      68           1 :         if rl.predicate != nil && !rl.predicate.Evaluate(op) {
      69           1 :                 return nil
      70           1 :         }
      71           1 :         var dur time.Duration
      72           1 :         rl.keyedPrng.withKey(op.Path, func(prng *rand.Rand) {
      73           1 :                 // We cap the max latency to 100x: Otherwise, it seems possible
      74           1 :                 // (although very unlikely) ExpFloat64 generates a multiplier high
      75           1 :                 // enough that causes a test timeout.
      76           1 :                 dur = time.Duration(min(prng.ExpFloat64(), 20.0) * float64(rl.mean))
      77           1 :         })
      78           1 :         time.Sleep(dur)
      79           1 :         return nil
      80             : }
      81             : 
      82             : // keyedPrng maintains a separate prng per-key that's deterministic with
      83             : // respect to the key: its behavior for a particular key is deterministic
      84             : // regardless of intervening evaluations for operations on other keys. This can
      85             : // be used to ensure determinism despite nondeterministic concurrency if the
      86             : // concurrency is constrained to separate keys.
      87             : type keyedPrng struct {
      88             :         rootSeed int64
      89             :         mu       struct {
      90             :                 sync.Mutex
      91             :                 h           maphash.Hash
      92             :                 perFilePrng map[string]*rand.Rand
      93             :         }
      94             : }
      95             : 
      96           1 : func (p *keyedPrng) init(rootSeed int64) {
      97           1 :         p.rootSeed = rootSeed
      98           1 :         p.mu.perFilePrng = make(map[string]*rand.Rand)
      99           1 : }
     100             : 
     101           1 : func (p *keyedPrng) withKey(key string, fn func(*rand.Rand)) {
     102           1 :         p.mu.Lock()
     103           1 :         defer p.mu.Unlock()
     104           1 :         prng, ok := p.mu.perFilePrng[key]
     105           1 :         if !ok {
     106           1 :                 // This is the first time an operation has been performed on the key.
     107           1 :                 // Initialize the per-key prng by computing a deterministic hash of the
     108           1 :                 // key.
     109           1 :                 p.mu.h.Reset()
     110           1 :                 var b [8]byte
     111           1 :                 binary.LittleEndian.PutUint64(b[:], uint64(p.rootSeed))
     112           1 :                 if _, err := p.mu.h.Write(b[:]); err != nil {
     113           0 :                         panic(err)
     114             :                 }
     115           1 :                 if _, err := p.mu.h.WriteString(key); err != nil {
     116           0 :                         panic(err)
     117             :                 }
     118           1 :                 seed := p.mu.h.Sum64()
     119           1 :                 prng = rand.New(rand.NewSource(int64(seed)))
     120           1 :                 p.mu.perFilePrng[key] = prng
     121             :         }
     122           1 :         fn(prng)
     123             : }