log n of p

cmd/.template/main.go

@@ -8,7 +8,7 @@ import (
 	"os"
 	"time"
 
-	"pg/src/with"
+	"gitea.bel.blue/bel/with"
 
 	_ "github.com/lib/pq"
 )

cmd/pg-fill/main.go

@@ -10,7 +10,7 @@ import (
 	"path"
 	"time"
 
-	"pg/src/with"
+	"gitea.bel.blue/bel/with"
 
 	_ "github.com/lib/pq"
 )

cmd/pg-lo-demo/main.go

@@ -8,7 +8,7 @@ import (
 	"fmt"
 	"os"
 
-	"pg/src/with"
+	"gitea.bel.blue/bel/with"
 
 	_ "github.com/lib/pq"
 )

cmd/pg-lockless-fifo-demo/main.go

@@ -13,7 +13,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"pg/src/with"
+	"gitea.bel.blue/bel/with"
 )
 
 func main() {

cmd/pg-pubsub-demo/main.go

@@ -11,7 +11,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"pg/src/with"
+	"gitea.bel.blue/bel/with"
 
 	_ "github.com/lib/pq"
 )

cmd/pg-pulse/main.go

@@ -7,9 +7,10 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"sync"
 	"time"
 
-	"pg/src/with"
+	"gitea.bel.blue/bel/with"
 
 	_ "github.com/lib/pq"
 )
@@ -24,16 +25,35 @@ 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++ {
+		log.Printf("dialing %v of %v background connections...", i+1, *p-1)
+		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";
-	   	CREATE TABLE IF NOT EXISTS "pg-pulse" (k TEXT);
-	   `); err != nil {
+			DROP TABLE IF EXISTS "pg-pulse";
+			CREATE TABLE IF NOT EXISTS "pg-pulse" (k TEXT);
+		`); err != nil {
 			return err
 		}
 
@@ -78,8 +98,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)
-	   	`)); err != nil {
+	  			INSERT INTO "pg-pulse" (k) SELECT substr(md5(random()::text), 1, 25)
+	  		`)); err != nil {
 				pushOK(false)
 				log.Println("\n! failed nonzero insert:", err)
 			} else {
@@ -89,22 +109,22 @@ func run(ctx context.Context) error {
 
 			ha, ha1 := false, false
 			row := pg.QueryRowContext(ctx, `
-	   		SELECT
-	   			(
-	   				SELECT COUNT(*) > 0
-	   				FROM pg_replication_slots
-	   				WHERE slot_name LIKE 'dpg_%_a_ha__'
-	   			) AS is_ha,
-	   			(
-	   				SELECT COUNT(*) > 0
-	   				FROM pg_replication_slots
-	   				WHERE slot_name LIKE 'dpg_%_a_ha_0'
-	   			) AS is_ha1,
-	   			(
-	   				SELECT $1 - COUNT(*)
-	   				FROM "pg-pulse"
-	   			) AS lost_writes
-	   	`, n)
+				SELECT
+					(
+						SELECT COUNT(*) > 0
+						FROM pg_replication_slots
+						WHERE slot_name LIKE 'dpg_%_a_ha__'
+					) AS is_ha,
+					(
+						SELECT COUNT(*) > 0
+						FROM pg_replication_slots
+						WHERE slot_name LIKE 'dpg_%_a_ha_0'
+					) AS is_ha1,
+					(
+						SELECT $1 - COUNT(*)
+						FROM "pg-pulse"
+					) AS lost_writes
+			`, 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 {

cmd/pg-queue-demo/main.go

@@ -11,7 +11,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"pg/src/with"
+	"gitea.bel.blue/bel/with"
 
 	_ "github.com/lib/pq"
 )

cmd/pg-spam/main.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"flag"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"sync/atomic"
+	"time"
+
+	"gitea.bel.blue/bel/with"
+
+	_ "github.com/lib/pq"
+)
+
+func main() {
+	if err := with.Context(run); err != nil {
+		panic(err)
+	}
+}
+
+func run(ctx context.Context) error {
+	fs := flag.NewFlagSet(os.Args[0], flag.ContinueOnError)
+	c := fs.String("c", "sqlite://", "conn string")
+	rps := fs.Int("rps", 1, "connections per second")
+	if err := fs.Parse(os.Args[1:]); err != nil {
+		panic(err)
+	}
+
+	return with.SQL(ctx, *c, func(pg *sql.DB) error {
+		stderr := log.Writer()
+		log.SetOutput(io.Discard)
+
+		var okays atomic.Uint64
+		var errs atomic.Uint64
+		d := time.Second / time.Duration(*rps)
+		go with.Every(ctx, 1*time.Second, func() {
+			fmt.Fprintf(stderr, "%v connections, %v failures\n", okays.Load(), errs.Load())
+		})
+		with.GoEvery(ctx, d, func() {
+			if err := with.SQL(ctx, *c, func(pg *sql.DB) error {
+				return nil
+			}); err != nil {
+				fmt.Fprintf(stderr, "err: %v\n", err)
+				errs.Add(1)
+			} else {
+				okays.Add(1)
+			}
+		})
+		return ctx.Err()
+	})
+}

go.mod

@@ -3,19 +3,20 @@ module pg
 go 1.24.2
 
 require (
-	github.com/lib/pq v1.10.9
-	modernc.org/sqlite v1.40.1
+	github.com/lib/pq v1.11.2
+	modernc.org/sqlite v1.46.1
 )
 
 require (
+	gitea.bel.blue/bel/with v0.0.0-20260309154209-886c4aabffff // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
-	github.com/ncruces/go-strftime v0.1.9 // indirect
+	github.com/ncruces/go-strftime v1.0.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
-	golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b // indirect
-	golang.org/x/sys v0.36.0 // indirect
-	modernc.org/libc v1.66.10 // indirect
+	golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
+	golang.org/x/sys v0.37.0 // indirect
+	modernc.org/libc v1.67.6 // indirect
 	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.11.0 // indirect
 )

go.sum

@@ -1,3 +1,5 @@
+gitea.bel.blue/bel/with v0.0.0-20260309154209-886c4aabffff h1:n5s/2L38pZBIIfaqBhSztaXVhDO0FerHl7MGMKOXUhM=
+gitea.bel.blue/bel/with v0.0.0-20260309154209-886c4aabffff/go.mod h1:VpayMFevpBM3hsEKxMPYyc3PGetoEhoshqs6mZD8Yec=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
@@ -6,14 +8,20 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/lib/pq v1.11.2 h1:x6gxUeu39V0BHZiugWe8LXZYZ+Utk7hSJGThs8sdzfs=
+github.com/lib/pq v1.11.2/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
 github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
+github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
+github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b h1:M2rDM6z3Fhozi9O7NWsxAkg/yqS/lQJ6PmkyIV3YP+o=
 golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b/go.mod h1:3//PLf8L/X+8b4vuAfHzxeRUl04Adcb341+IGKfnqS8=
+golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 h1:mgKeJMpvi0yx/sU5GsxQ7p6s2wtOnGAHZWCHUM4KGzY=
+golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546/go.mod h1:j/pmGrbnkbPtQfxEe5D0VQhZC6qKbfKifgD0oM7sR70=
 golang.org/x/mod v0.27.0 h1:kb+q2PyFnEADO2IEF935ehFUXlWiNjJWtRNgBLSfbxQ=
 golang.org/x/mod v0.27.0/go.mod h1:rWI627Fq0DEoudcK+MBkNkCe0EetEaDSwJJkCcjpazc=
 golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw=
@@ -21,6 +29,8 @@ golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.36.0 h1:KVRy2GtZBrk1cBYA7MKu5bEZFxQk4NIDV6RLVcC8o0k=
 golang.org/x/sys v0.36.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
+golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/tools v0.36.0 h1:kWS0uv/zsvHEle1LbV5LE8QujrxB3wfQyxHfhOk0Qkg=
 golang.org/x/tools v0.36.0/go.mod h1:WBDiHKJK8YgLHlcQPYQzNCkUxUypCaa5ZegCVutKm+s=
 modernc.org/cc/v4 v4.26.5 h1:xM3bX7Mve6G8K8b+T11ReenJOT+BmVqQj0FY5T4+5Y4=
@@ -35,6 +45,8 @@ modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks=
 modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI=
 modernc.org/libc v1.66.10 h1:yZkb3YeLx4oynyR+iUsXsybsX4Ubx7MQlSYEw4yj59A=
 modernc.org/libc v1.66.10/go.mod h1:8vGSEwvoUoltr4dlywvHqjtAqHBaw0j1jI7iFBTAr2I=
+modernc.org/libc v1.67.6 h1:eVOQvpModVLKOdT+LvBPjdQqfrZq+pC39BygcT+E7OI=
+modernc.org/libc v1.67.6/go.mod h1:JAhxUVlolfYDErnwiqaLvUqc8nfb2r6S6slAgZOnaiE=
 modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
 modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
 modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
@@ -45,6 +57,8 @@ modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
 modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
 modernc.org/sqlite v1.40.1 h1:VfuXcxcUWWKRBuP8+BR9L7VnmusMgBNNnBYGEe9w/iY=
 modernc.org/sqlite v1.40.1/go.mod h1:9fjQZ0mB1LLP0GYrp39oOJXx/I2sxEnZtzCmEQIKvGE=
+modernc.org/sqlite v1.46.1 h1:eFJ2ShBLIEnUWlLy12raN0Z1plqmFX9Qe3rjQTKt6sU=
+modernc.org/sqlite v1.46.1/go.mod h1:CzbrU2lSB1DKUusvwGz7rqEKIq+NUd8GWuBBZDs9/nA=
 modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
 modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
 modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=

src/with/ctx.go

@@ -1,17 +0,0 @@
-package with
-
-import (
-	"context"
-	"os/signal"
-	"syscall"
-)
-
-func Context(foo func(context.Context) error) error {
-	ctx, can := signal.NotifyContext(context.Background(), syscall.SIGINT)
-	defer can()
-
-	if err := foo(ctx); err != nil && ctx.Err() == nil {
-		return err
-	}
-	return nil
-}

src/with/every.go

@@ -1,34 +0,0 @@
-package with
-
-import (
-	"context"
-	"time"
-)
-
-func GoEvery(ctx context.Context, d time.Duration, foo func()) {
-	every(ctx, d, foo, true)
-}
-
-func Every(ctx context.Context, d time.Duration, foo func()) {
-	every(ctx, d, foo, false)
-}
-
-func every(ctx context.Context, d time.Duration, foo func(), async bool) {
-	ticker := time.NewTicker(d)
-	defer ticker.Stop()
-	for ctx.Err() == nil {
-		everyTry(ctx, foo, async)
-		select {
-		case <-ctx.Done():
-		case <-ticker.C:
-		}
-	}
-}
-
-func everyTry(ctx context.Context, foo func(), async bool) {
-	if async {
-		go foo()
-	} else {
-		foo()
-	}
-}

src/with/pg.go

@@ -1,12 +0,0 @@
-package with
-
-import (
-	"context"
-	"database/sql"
-
-	_ "github.com/lib/pq"
-)
-
-func PSQL(ctx context.Context, conn string, foo func(db *sql.DB) error) error {
-	return _sql(ctx, "postgres", conn, foo)
-}

src/with/sql.go

@@ -1,71 +0,0 @@
-package with
-
-import (
-	"context"
-	"database/sql"
-	"fmt"
-	"log"
-	"net/url"
-	"time"
-
-	_ "modernc.org/sqlite"
-)
-
-func SQL(ctx context.Context, conn string, foo func(*sql.DB) error) error {
-	u, err := url.Parse(conn)
-	if err != nil {
-		return err
-	}
-	switch u.Scheme {
-	case "sqlite":
-		return Sqlite(ctx, conn, foo)
-	case "postgres", "postgresql":
-		return PSQL(ctx, conn, foo)
-	}
-	return fmt.Errorf("unknown sql scheme %q", u.Scheme)
-}
-
-func _sql(ctx context.Context, engine, conn string, foo func(db *sql.DB) error) error {
-	log.Printf("opening %s %s...", engine, conn)
-	db, err := sql.Open(engine, conn)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		log.Println("closed:", db.Close())
-	}()
-
-	func() {
-		pinged := make(chan bool)
-		defer close(pinged)
-		for {
-			log.Println("pinging...")
-			go func() {
-				err := db.PingContext(ctx)
-				if err != nil {
-					log.Println("!", err)
-				}
-				select {
-				case pinged <- err == nil:
-				case <-ctx.Done():
-				case <-time.After(time.Second * 5):
-				}
-			}()
-			select {
-			case <-ctx.Done():
-				return
-			case ok := <-pinged:
-				if ok {
-					return
-				}
-			}
-			select {
-			case <-ctx.Done():
-			case <-time.After(time.Second):
-			}
-		}
-	}()
-	log.Println("connected")
-
-	return foo(db)
-}

src/with/sqlite.go

@@ -1,14 +0,0 @@
-package with
-
-import (
-	"context"
-	"database/sql"
-	"strings"
-
-	_ "modernc.org/sqlite"
-)
-
-func Sqlite(ctx context.Context, conn string, foo func(db *sql.DB) error) error {
-	conn = strings.TrimPrefix(conn, "sqlite://")
-	return _sql(ctx, "sqlite", conn, foo)
-}