wip

.gitignore

@@ -1 +1,4 @@
 **/*.sw*
+/cmd/pg-lo-demo/pg-lo-demo
+/cmd/pg-pulse/pg-pulse
+/cmd/pg-walspam/pg-walspam

cmd/pg-lo-demo/main.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"flag"
+	"os"
+
+	"pg/src/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", "postgresql://pulsegres:pulsegres@localhost:15432", "conn string")
+	if err := fs.Parse(os.Args[1:]); err != nil {
+		panic(err)
+	}
+
+	return with.PSQL(ctx, *c, func(pg *sql.DB) error {
+		// SELECT lo_creat(-1);
+		// SELECT pg_catalog.lowrite(0, '\\037\\213\\010\\000\\000\\000\\000\\000\\000\\000\\245\\220;n\\3030\\014')
+		return ctx.Err()
+	})
+}

cmd/pg-pulse/main.go

@@ -7,17 +7,20 @@ import (
 	"fmt"
 	"log"
 	"os"
-	"os/signal"
-	"syscall"
 	"time"
 
+	"pg/src/with"
+
 	_ "github.com/lib/pq"
 )
 
 func main() {
-	ctx, can := signal.NotifyContext(context.Background(), syscall.SIGINT)
+	if err := with.Context(run); err != nil {
-	defer can()
+		panic(err)
+	}
+}
 
+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")
@@ -25,154 +28,92 @@ func main() {
 		panic(err)
 	}
 
-	log.Println("opening...")
+	return with.PSQL(ctx, *c, func(pg *sql.DB) error {
-	pg, err := sql.Open("postgres", *c)
+		log.Println("staging...")
-	if err != nil {
+		if _, err := pg.ExecContext(ctx, `
-		panic(err)
+	   	DROP TABLE IF EXISTS "pg-pulse";
-	}
+	   	CREATE TABLE IF NOT EXISTS "pg-pulse" (k TEXT);
-	defer func() {
+	   `); err != nil {
-		log.Println("closed:", pg.Close())
+			return err
-	}()
+		}
 
-	func() {
+		log.Println("spamming...")
-		pinged := make(chan bool)
+
-		defer close(pinged)
+		var downtime time.Duration
-		for {
+		okc := make(chan bool)
-			log.Println("pinging...")
+		defer close(okc)
-			go func() {
+		go func() {
-				err := pg.PingContext(ctx)
+			lastOK := true
-				if err != nil {
+			wasOK := time.Now()
-					log.Println("!", err)
+			for ok := range okc {
+				if isNewlyOK := ok && !lastOK; isNewlyOK {
+					downtime += time.Since(wasOK)
 				}
-				select {
+				lastOK = ok
-				case pinged <- err == nil:
-				case <-ctx.Done():
-				case <-time.After(time.Second * 5):
-				}
-			}()
-			select {
-			case <-ctx.Done():
-				return
-			case ok := <-pinged:
 				if ok {
-					return
+					wasOK = time.Now()
 				}
-			case <-time.After(time.Second * 5):
-			}
-		}
-	}()
-
-	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 {
-		panic(err)
-	}
-
-	log.Println("spamming...")
-
-	var downtime time.Duration
-	okc := make(chan bool)
-	defer close(okc)
-	go func() {
-		lastOK := true
-		wasOK := time.Now()
-		for ok := range okc {
-			if isNewlyOK := ok && !lastOK; isNewlyOK {
-				downtime += time.Since(wasOK)
-			}
-			lastOK = ok
-			if ok {
-				wasOK = time.Now()
-			}
-		}
-	}()
-	pushOK := func(v bool) {
-		select {
-		case okc <- v:
-		default:
-		}
-	}
-
-	pticker := time.NewTicker(5 * time.Second)
-	defer pticker.Stop()
-	ticker := time.NewTicker(*d)
-	defer ticker.Stop()
-	n := uint(0)
-	isHA := false
-	isHA0 := false
-	for ctx.Err() == nil {
-		func() {
-			ctx, can := context.WithCancel(ctx)
-			defer can()
-
-			done := make(chan struct{})
-			go func() {
-				defer close(done)
-
-				if _, err := pg.ExecContext(ctx, fmt.Sprintf(`
-					INSERT INTO "pg-pulse" (k) SELECT substr(md5(random()::text), 1, 25)
-				`)); err != nil {
-					pushOK(false)
-					log.Println("\n! failed nonzero insert:", err)
-				} else {
-					n += 1
-					pushOK(true)
-				}
-
-				var lostWrites uint
-				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)
-				if err := row.Err(); err != nil {
-					log.Println("\n! failed getting ha-stuff:", err)
-				} else if err := row.Scan(&ha, &ha1, &lostWrites); err != nil {
-					log.Println("\n! failed scanning ha-stuff:", err)
-				} else {
-					isHA = isHA || ha
-					isHA0 = isHA && !ha1
-				}
-
-				ctx, can = context.WithTimeout(ctx, time.Second*5)
-				defer can()
-
-				select {
-				case <-pticker.C:
-					using := ""
-					if isHA && isHA0 {
-						using = "using ha0"
-					} else if isHA && !isHA0 {
-						using = "using ha1"
-					}
-					log.Printf("%v writes (%v lost) (%v down) %s", n, lostWrites, downtime.Round(time.Second), using)
-				default:
-				}
-			}()
-
-			select {
-			case <-ctx.Done():
-			case <-done:
 			}
 		}()
-		select {
+		pushOK := func(v bool) {
-		case <-ctx.Done():
+			select {
-		case <-ticker.C:
+			case okc <- v:
+			default:
+			}
 		}
-	}
+
+		n := uint(0)
+		isHA := false
+		isHA0 := false
+		var lostWrites uint
+		go with.Every(ctx, 5*time.Second, func() {
+			using := ""
+			if isHA && isHA0 {
+				using = "using ha0"
+			} else if isHA && !isHA0 {
+				using = "using ha1"
+			}
+			log.Printf("%v writes (%v lost) (%v down) %s", n, lostWrites, downtime.Round(time.Second), using)
+		})
+
+		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 {
+				pushOK(false)
+				log.Println("\n! failed nonzero insert:", err)
+			} else {
+				n += 1
+				pushOK(true)
+			}
+
+			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)
+			if err := row.Err(); err != nil {
+				log.Println("\n! failed getting ha-stuff:", err)
+			} else if err := row.Scan(&ha, &ha1, &lostWrites); err != nil {
+				log.Println("\n! failed scanning ha-stuff:", err)
+			} else {
+				isHA = isHA || ha
+				isHA0 = isHA && !ha1
+			}
+		})
+		return ctx.Err()
+	})
 }

src/with/ctx.go

@@ -0,0 +1,17 @@
+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

@@ -0,0 +1,34 @@
+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

@@ -0,0 +1,51 @@
+package with
+
+import (
+	"context"
+	"database/sql"
+	"log"
+	"time"
+
+	_ "github.com/lib/pq"
+)
+
+func PSQL(ctx context.Context, conn string, foo func(db *sql.DB) error) error {
+	log.Println("opening...")
+	pg, err := sql.Open("postgres", conn)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		log.Println("closed:", pg.Close())
+	}()
+
+	func() {
+		pinged := make(chan bool)
+		defer close(pinged)
+		for {
+			log.Println("pinging...")
+			go func() {
+				err := pg.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
+				}
+			case <-time.After(time.Second * 5):
+			}
+		}
+	}()
+
+	return foo(pg)
+}