Function TestConnConcurrency

stdlib/sql_test.go:335–421

(t *testing.T)

}

func TestConnConcurrency(t *testing.T) {
	testWithAllQueryExecModes(t, func(t *testing.T, db *sql.DB) {
		_, err := db.Exec("create table t (id integer primary key, str text, dur_str interval)")
		require.NoError(t, err)

		defer func() {
			_, err := db.Exec("drop table t")
			require.NoError(t, err)
		}()

		var wg sync.WaitGroup

		concurrency := 50
		errChan := make(chan error, concurrency)

		for i := 1; i <= concurrency; i++ {
			wg.Go(func() {
				ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
				defer cancel()

				str := strconv.Itoa(i)
				duration := time.Duration(i) * time.Second
				_, err := db.ExecContext(ctx, "insert into t values($1)", i)
				if err != nil {
					errChan <- fmt.Errorf("insert failed: %d %w", i, err)
					return
				}
				_, err = db.ExecContext(ctx, "update t set str = $1 where id = $2", str, i)
				if err != nil {
					errChan <- fmt.Errorf("update 1 failed: %d %w", i, err)
					return
				}
				_, err = db.ExecContext(ctx, "update t set dur_str = $1 where id = $2", duration, i)
				if err != nil {
					errChan <- fmt.Errorf("update 2 failed: %d %w", i, err)
					return
				}

				errChan <- nil
			})
		}
		wg.Wait()
		for i := 1; i <= concurrency; i++ {
			err := <-errChan
			require.NoError(t, err)
		}

		for i := 1; i <= concurrency; i++ {
			wg.Go(func() {
				ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
				defer cancel()

				var id int
				var str string
				var duration pgtype.Interval
				err := db.QueryRowContext(ctx, "select id,str,dur_str from t where id = $1", i).Scan(&id, &str, &duration)
				if err != nil {
					errChan <- fmt.Errorf("select failed: %d %w", i, err)