Method run

concurrent_test.go:350–393

()

}

func (w *worker) run() (historyRecords, error) {
	var rs historyRecords

	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-w.stopCh:
			return rs, nil
		default:
		}

		err := w.db.Update(func(tx *bolt.Tx) error {
			for {
				op := w.pickOperation()
				bucket, key := w.pickBucket(), w.pickKey()
				rec, eerr := executeOperation(op, tx, bucket, key, w.conf)
				if eerr != nil {
					opErr := fmt.Errorf("[%s: %s]: %w", w.name(), op, eerr)
					w.t.Error(opErr)
					w.errCh <- opErr
					return opErr
				}

				rs = append(rs, rec)
				if w.conf.workInterval != (duration{}) {
					time.Sleep(randomDurationInRange(w.conf.workInterval.min, w.conf.workInterval.max))
				}

				select {
				case <-ticker.C:
					return nil
				case <-w.stopCh:
					return nil
				default:
				}
			}
		})
		if err != nil {
			return rs, err
		}
	}
}

func (w *worker) pickBucket() []byte {
	return bucketName(mrand.Intn(w.conf.bucketCount))