open 1 conn at a time

cmd/pg-pulse/main.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"sync"
 	"time"
 
 	"gitea.bel.blue/bel/with"
@@ -24,16 +25,34 @@ func run(ctx context.Context) error {
 	fs := flag.NewFlagSet(os.Args[0], flag.ContinueOnError)
 	c := fs.String("c", "postgresql://pulsegres:pulsegres@localhost:15432", "conn string")
 	d := fs.Duration("d", time.Second, "interval")
+	p := fs.Int("p", 1, "concurrent goroutines")
 	if err := fs.Parse(os.Args[1:]); err != nil {
 		panic(err)
 	}
 
+	for i := 0; i < *p-1; i++ {
+		func() {
+			connected := &sync.WaitGroup{}
+			connected.Add(1)
+			go with.PSQL(ctx, *c, func(pg *sql.DB) error {
+				connected.Done()
+				with.GoEvery(ctx, *d, func() {
+					if _, err := pg.ExecContext(ctx, `SELECT 1`); err != nil {
+						log.Println("!", err)
+					}
+				})
+				return ctx.Err()
+			})
+			connected.Wait()
+		}()
+	}
+
 	return with.PSQL(ctx, *c, func(pg *sql.DB) error {
 		log.Println("staging...")
 		if _, err := pg.ExecContext(ctx, `
-	   	DROP TABLE IF EXISTS "pg-pulse";
+			DROP TABLE IF EXISTS "pg-pulse";
-	   	CREATE TABLE IF NOT EXISTS "pg-pulse" (k TEXT);
+			CREATE TABLE IF NOT EXISTS "pg-pulse" (k TEXT);
-	   `); err != nil {
+		`); err != nil {
 			return err
 		}
 
@@ -78,8 +97,8 @@ func run(ctx context.Context) error {
 
 		with.GoEvery(ctx, *d, func() {
 			if _, err := pg.ExecContext(ctx, fmt.Sprintf(`
-	   		INSERT INTO "pg-pulse" (k) SELECT substr(md5(random()::text), 1, 25)
+	  			INSERT INTO "pg-pulse" (k) SELECT substr(md5(random()::text), 1, 25)
-	   	`)); err != nil {
+	  		`)); err != nil {
 				pushOK(false)
 				log.Println("\n! failed nonzero insert:", err)
 			} else {
@@ -89,22 +108,22 @@ func run(ctx context.Context) error {
 
 			ha, ha1 := false, false
 			row := pg.QueryRowContext(ctx, `
-	   		SELECT
+				SELECT
-	   			(
+					(
-	   				SELECT COUNT(*) > 0
+						SELECT COUNT(*) > 0
-	   				FROM pg_replication_slots
+						FROM pg_replication_slots
-	   				WHERE slot_name LIKE 'dpg_%_a_ha__'
+						WHERE slot_name LIKE 'dpg_%_a_ha__'
-	   			) AS is_ha,
+					) AS is_ha,
-	   			(
+					(
-	   				SELECT COUNT(*) > 0
+						SELECT COUNT(*) > 0
-	   				FROM pg_replication_slots
+						FROM pg_replication_slots
-	   				WHERE slot_name LIKE 'dpg_%_a_ha_0'
+						WHERE slot_name LIKE 'dpg_%_a_ha_0'
-	   			) AS is_ha1,
+					) AS is_ha1,
-	   			(
+					(
-	   				SELECT $1 - COUNT(*)
+						SELECT $1 - COUNT(*)
-	   				FROM "pg-pulse"
+						FROM "pg-pulse"
-	   			) AS lost_writes
+					) AS lost_writes
-	   	`, n)
+			`, n)
 			if err := row.Err(); err != nil {
 				log.Println("\n! failed getting ha-stuff:", err)
 			} else if err := row.Scan(&ha, &ha1, &lostWrites); err != nil {