db_test.go

package dbtest

import (
	"database/sql"
	"fmt"
	"os"
	"sync"
	"testing"
	"time"

	_ "github.com/mattn/go-sqlite3"
)

func TestConcurrency(t *testing.T) {
	os.Remove("test.db")
	os.Remove("test.db-shm")
	os.Remove("test.db-wal")

	db, err := sql.Open("sqlite3", "test.db")
	if err != nil {
		t.Fatal(err)
	}

	_, err = db.Exec("PRAGMA journal_mode=WAL")
	if err != nil {
		t.Fatal(err)
	}
	_, err = db.Exec("CREATE TABLE test (id SERIAL, user TEXT NOT NULL, name TEXT NOT NULL)")
	if err != nil {
		t.Fatal(err)
	}
	_, err = db.Exec("INSERT INTO test (user, name) VALUES ('test1','test1')")
	if err != nil {
		t.Fatal(err)
	}

	t0 := time.Now()
	secs := 0
	iterations := 0
	for time.Since(t0) < 10*time.Second {
		if s := int(time.Since(t0).Seconds()); s != secs {
			fmt.Println(iterations, "iterations,", s, "seconds...")
			secs = s
			iterations = 0
		}
		var wg sync.WaitGroup
		start := make(chan struct{})

		wg.Add(1)
		go func() {
			t0 := time.Now()
			insertOne(start, t, db)
			if since := time.Since(t0); since > 10*time.Millisecond {
				fmt.Println("write took", time.Since(t0))
			}
			wg.Done()
		}()

		wg.Add(1)
		go func() {
			t0 := time.Now()
			insertOne(start, t, db)
			if since := time.Since(t0); since > 10*time.Millisecond {
				fmt.Println("write took", time.Since(t0))
			}
			wg.Done()
		}()

		wg.Add(1)
		go func() {
			t0 := time.Now()
			readOne(start, t, db)
			if since := time.Since(t0); since > 10*time.Millisecond {
				fmt.Println("read took", time.Since(t0))
			}
			wg.Done()
		}()

		wg.Add(1)
		go func() {
			t0 := time.Now()
			readOne(start, t, db)
			if since := time.Since(t0); since > 10*time.Millisecond {
				fmt.Println("read took", time.Since(t0))
			}
			wg.Done()
		}()

		close(start)
		wg.Wait()
		iterations++
	}
}

func insertOne(start chan struct{}, t *testing.T, db *sql.DB) {
	trans, err := db.Begin()
	if err != nil {
		t.Fatal(err)
	}

	<-start

	_, err = trans.Exec("INSERT INTO test (user, name) VALUES ('test2', 'test2')")
	if err != nil {
		t.Fatal(err)
	}

	_, err = trans.Exec("DELETE FROM test WHERE user = 'test2'")
	if err != nil {
		t.Fatal(err)
	}

	err = trans.Commit()
	if err != nil {
		t.Fatal(err)
	}
}

func readOne(start chan struct{}, t *testing.T, db *sql.DB) {
	trans, err := db.Begin()
	if err != nil {
		fmt.Printf("begin: %v\n", err)
	}
	defer trans.Rollback()

	rows, err := trans.Query("SELECT user, name FROM test")
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()

	<-start

	count := 0
	for rows.Next() {
		var user, name string
		if err := rows.Scan(&user, &name); err != nil {
			t.Fatal(err)
		}
		count++
	}

	if count != 1 {
		fmt.Println("got", count, "rows")
	}
}

$ go test -v -cpu 4
=== RUN   TestConcurrency
write took 17.242675ms
write took 1.017988063s
1 iterations, 1 seconds...
write took 17.360597ms
write took 1.015084445s
1 iterations, 2 seconds...
write took 15.188925ms
write took 1.013220409s
1 iterations, 3 seconds...
write took 11.576886ms
write took 1.011122606s
1 iterations, 4 seconds...
write took 11.478028ms
write took 1.017511937s
1 iterations, 5 seconds...
write took 16.899053ms
write took 1.016841776s
1 iterations, 6 seconds...
write took 15.0625ms
write took 1.015102284s
1 iterations, 7 seconds...
write took 12.465045ms
write took 1.011578382s
1 iterations, 8 seconds...
write took 11.425128ms
write took 1.011647859s
1 iterations, 9 seconds...
write took 11.619235ms
write took 1.014100957s
--- PASS: TestConcurrency (10.24s)
PASS
ok      github.com  10.244s