package oltp
import (
"context"
"fmt"
"sync"
"testing"
)
// Ticket numbering reads the repository's highest number and then
// inserts above it inside one transaction. Many of those running at
// once must all commit: a transaction that began with a read and then
// found another writer had landed first used to fail with
// SQLITE_BUSY_SNAPSHOT instead of waiting its turn.
func TestConcurrentReadThenWriteTransactionsAllCommit(t *testing.T) {
d := openTest(t)
const workers, rounds = 8, 25
var wg sync.WaitGroup
errs := make(chan error, workers*rounds)
for w := 0; w < workers; w++ {
wg.Add(1)
go func(w int) {
defer wg.Done()
for r := 0; r < rounds; r++ {
if err := d.TicketNumbersAssign("q", []string{fmt.Sprintf("w%d-r%d", w, r)}); err != nil {
errs <- err
}
}
}(w)
}
wg.Wait()
close(errs)
failed := 0
for err := range errs {
if failed == 0 {
t.Errorf("FAIL: concurrent read-then-write transaction: %v", err)
}
failed++
}
if failed > 0 {
t.Fatalf("FAIL: %d of %d transactions failed", failed, workers*rounds)
}
got, err := d.TicketNumbers("q")
if err != nil {
t.Fatal(err)
}
seen := map[int]bool{}
for _, n := range got {
seen[n] = true
}
if len(got) != workers*rounds || len(seen) != workers*rounds {
t.Fatalf("FAIL: %d tickets numbered with %d distinct numbers, want %d", len(got), len(seen), workers*rounds)
}
t.Logf("ok: %d concurrent read-then-write transactions committed with distinct numbers", workers*rounds)
}
// Pragmas that live on a connection rather than in the file must be on
// every connection the pool opens, not only the first one.
func TestEveryPooledConnectionCarriesThePragmas(t *testing.T) {
d := openTest(t)
ctx := context.Background()
first, err := d.db.Conn(ctx)
if err != nil {
t.Fatal(err)
}
defer first.Close()
// Holding the first connection forces the pool to open another.
second, err := d.db.Conn(ctx)
if err != nil {
t.Fatal(err)
}
defer second.Close()
var busy int
var journal string
if err := second.QueryRowContext(ctx, `PRAGMA busy_timeout`).Scan(&busy); err != nil {
t.Fatal(err)
}
if err := second.QueryRowContext(ctx, `PRAGMA journal_mode`).Scan(&journal); err != nil {
t.Fatal(err)
}
if busy != 5000 || journal != "wal" {
t.Fatalf("FAIL: second connection has busy_timeout=%d journal_mode=%s, want 5000 and wal", busy, journal)
}
t.Logf("ok: second connection has busy_timeout=%d journal_mode=%s", busy, journal)
}
// A transaction begun on the reader pool must stay a reader: it may not
// take the write lock at BEGIN, or it would stall the writer queue and
// fail a writer that finds the lock held past the busy timeout.
func TestAReaderPoolTransactionDoesNotBlockTheWriter(t *testing.T) {
d := openTest(t)
if err := d.TicketNumbersAssign("q", []string{"aa"}); err != nil {
t.Fatal(err)
}
rtx, err := d.db.Begin()
if err != nil {
t.Fatal(err)
}
defer rtx.Rollback()
var n int
if err := rtx.QueryRow(`SELECT COUNT(*) FROM ticket_number`).Scan(&n); err != nil {
t.Fatal(err)
}
if err := d.TicketNumbersAssign("q", []string{"bb"}); err != nil {
t.Fatalf("FAIL: a writer transaction failed while a reader-pool transaction was open: %v", err)
}
t.Logf("ok: writer committed while a reader-pool transaction held a snapshot of %d rows", n)
}