repro.go

package main

import (
	"context"
	"database/sql"
	"flag"
	"fmt"
	"log"
	"sync"
	"time"

	_ "github.com/lib/pq"
)

var jitter = flag.Int("jitter-ms", 10, "jitter in millis")
var timeout = flag.Int("timeout-ms", 1000, "query timeout in millis")
var reps = flag.Int("reps", 100, "number of times the query is made")

func main() {
	flag.Parse()

	query := fmt.Sprintf(`SELECT pg_sleep_for('%d ms') || 'a string to return'`, *timeout)
	fmt.Println(query)

	var timeouts []time.Duration
	for i := 0; i < *jitter*1000; i++ {
		millis := time.Duration(*timeout-i) * time.Millisecond
		for j := 0; j < 100; j++ {
			timeouts = append(timeouts, millis-(time.Duration(j)*10*time.Microsecond))
		}
	}

	db, openErr := openAndPing("postgres", "postgres:///pq_test?sslmode=disable")
	if openErr != nil {
		log.Fatalf("failed to open a connection: %s", openErr)
	}

	_, createErr := db.Exec(`CREATE TABLE IF NOT EXISTS test (something text)`)
	if createErr != nil {
		log.Fatalf("failed to create table: %s", createErr)
	}

	for _, timeout := range timeouts {
		wg := sync.WaitGroup{}

		for i := 0; i < *reps; i++ {

			wg.Add(1)
			go func(rep int, timeout time.Duration) {
				defer wg.Done()

				ctx, cancel := context.WithTimeout(context.Background(), timeout)
				defer cancel()

				tx, err := db.BeginTx(ctx, nil)
				if err != nil {
					log.Printf("timeout=%.08fs rep=%d status=error error=%q", timeout.Seconds(), rep, err)
					return
				}

				var result string
				if err := tx.QueryRowContext(ctx, query).Scan(&result); err != nil {
					log.Printf("timeout=%.08fs rep=%d status=error error=%q", timeout.Seconds(), rep, err)
				} else {
					log.Printf("timeout=%.08fs rep=%d status=ok", timeout.Seconds(), rep)
				}

			}(i, timeout)
		}

		wg.Wait()
		time.Sleep(1 * time.Second)
	}
}

func openAndPing(driver, source string) (*sql.DB, error) {
	db, err := sql.Open("postgres", "postgres:///pq_test?sslmode=disable")
	if err != nil {
		return nil, err
	}

	if err := db.Ping(); err != nil {
		return nil, err
	}
	return db, nil
}