Method generateLoad

cmd/litestream-test/load.go:77–122

(ctx context.Context)

}

func (c *LoadCommand) generateLoad(ctx context.Context) error {
	db, err := sql.Open("sqlite3", c.DB+"?_journal_mode=WAL")
	if err != nil {
		return fmt.Errorf("open database: %w", err)
	}
	defer db.Close()

	db.SetMaxOpenConns(c.Workers + 1)
	db.SetMaxIdleConns(c.Workers)

	if err := c.ensureTestTable(db); err != nil {
		return fmt.Errorf("ensure test table: %w", err)
	}

	ctx, cancel := context.WithTimeout(ctx, c.Duration)
	defer cancel()

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
	go func() {
		<-sigChan
		slog.Info("Received interrupt signal, stopping load generation")
		cancel()
	}()

	stats := &LoadStats{
		startTime:  time.Now(),
		lastReport: time.Now(),
	}

	var wg sync.WaitGroup
	for i := 0; i < c.Workers; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			c.worker(ctx, db, workerID, stats)
		}(i)
	}

	go c.reportStats(ctx, stats)

	wg.Wait()

	c.finalReport(stats)
	return nil
}

func (c *LoadCommand) worker(ctx context.Context, db *sql.DB, workerID int, stats *LoadStats) {
	ticker := time.NewTicker(time.Second / time.Duration(c.WriteRate/c.Workers))