all: use core/async where approp

Author: ignoramous

intra/bootstrap.go

@@ -16,6 +16,7 @@ import (
 	"sync"
 
 	x "github.com/celzero/firestack/intra/backend"
+	"github.com/celzero/firestack/intra/core"
 	"github.com/celzero/firestack/intra/dns53"
 	"github.com/celzero/firestack/intra/dnsx"
 	"github.com/celzero/firestack/intra/doh"
@@ -233,7 +234,7 @@ func (b *bootstrap) kickstartLocked(px ipn.ProxyProvider) error {
 	}
 
 	if prev := b.tr; prev != nil {
-		go stopTransport(prev) // stop after new transport is ready
+		core.Gx1("dns.bootstrap.stop", stopTransport, prev) // stop after new transport is ready
 		log.I("dns: default: removing %s %s[%s]; using %s %s",
 			b.typ, b.hostname, b.IPPorts(), typstr(tr), ippstr(tr))
 	}

intra/common.go

@@ -12,6 +12,7 @@ import (
 	"math/rand"
 	"net"
 	"net/netip"
+	"runtime/debug"
 	"strconv"
 	"strings"
 	"sync"
@@ -439,6 +440,7 @@ func (h *baseHandler) End() {
 
 // TODO: Propagate TCP RST using local.Abort(), on appropriate errors.
 func upload(id string, local, remote net.Conn, ioch chan<- ioinfo) {
+	debug.SetPanicOnFault(true)
 	defer core.Recover(core.Exit11, "c.upload."+id)
 	defer core.CloseOp(local, core.CopR)
 	defer core.CloseOp(remote, core.CopW)

intra/core/async.go

@@ -9,13 +9,15 @@ package core
 import (
 	"context"
 	"errors"
+	"runtime/debug"
 	"strconv"
 	"time"
 )
 
 // Go runs f in a goroutine and recovers from any panics.
 func Go(who string, f func()) {
 	go func() {
+		debug.SetPanicOnFault(true)
 		defer Recover(DontExit, who)
 
 		f()
@@ -25,6 +27,7 @@ func Go(who string, f func()) {
 // Go1 runs f(arg) in a goroutine and recovers from any panics.
 func Go1[T any](who string, f func(T), arg T) {
 	go func() {
+		debug.SetPanicOnFault(true)
 		defer Recover(DontExit, who)
 
 		f(arg)
@@ -34,6 +37,7 @@ func Go1[T any](who string, f func(T), arg T) {
 // Go2 runs f(arg0,arg1) in a goroutine and recovers from any panics.
 func Go2[T0 any, T1 any](who string, f func(T0, T1), a0 T0, a1 T1) {
 	go func() {
+		debug.SetPanicOnFault(true)
 		defer Recover(DontExit, who)
 
 		f(a0, a1)
@@ -44,6 +48,7 @@ func Go2[T0 any, T1 any](who string, f func(T0, T1), a0 T0, a1 T1) {
 // then calls cb in a separate goroutine, and recovers from any panics.
 func Gg(who string, f func(), cb func()) {
 	go func() {
+		debug.SetPanicOnFault(true)
 		defer RecoverFn(who, cb)
 
 		f()
@@ -53,12 +58,23 @@ func Gg(who string, f func(), cb func()) {
 // Gx runs f in a goroutine and exits the process if f panics.
 func Gx(who string, f func()) {
 	go func() {
+		debug.SetPanicOnFault(true)
 		defer Recover(Exit11, who)
 
 		f()
 	}()
 }
 
+// Gx1 runs f in a goroutine and exits the process if f panics.
+func Gx1[T any](who string, f func(T), arg T) {
+	go func() {
+		debug.SetPanicOnFault(true)
+		defer Recover(Exit11, who)
+
+		f(arg)
+	}()
+}
+
 // Gif runs f in a goroutine if cond is true.
 func Gif(cond bool, who string, f func()) {
 	if cond {
@@ -75,6 +91,7 @@ func Grx[T any](who string, f WorkCtx[T], d time.Duration) (zz T, completed bool
 
 	// go.dev/play/p/VtWYJrxhXz6
 	go func() {
+		debug.SetPanicOnFault(true)
 		defer Recover(Exit11, who)
 		defer close(ch)
 
@@ -90,6 +107,16 @@ func Grx[T any](who string, f WorkCtx[T], d time.Duration) (zz T, completed bool
 	return zz, false
 }
 
+// Gxe runs f in a goroutine, ignores returned error, and exits on panics.
+func Gxe(who string, f func() error) {
+	go func() {
+		debug.SetPanicOnFault(true)
+		defer Recover(Exit11, who)
+
+		_ = f()
+	}()
+}
+
 // errPanic returns an error indicating that the function at index i panicked.
 func errPanic(who string) error {
 	return errors.New(who + " fn panicked")
@@ -154,7 +181,7 @@ loop:
 func First[T any](who string, overallTimeout time.Duration, fs ...WorkCtx[T]) (zz T, idx int) {
 	timeoutPerFn := overallTimeout / time.Duration(len(fs))
 	for i, f := range fs {
-		i, f := i, f
+		// unneeded in go1.23+ i, f := i, f
 		fid := who + ".all." + strconv.Itoa(i)
 		if x, ok := Grx(fid, f, timeoutPerFn); ok {
 			return x, i
@@ -176,7 +203,7 @@ func All[T any](who string, timeout time.Duration, fs ...WorkCtx[T]) ([]T, []err
 	defer cancel()
 
 	for i, f := range fs {
-		i, f := i, f
+		//unneeded in go1.23+ i, f := i, f
 		fid := who + ".all." + strconv.Itoa(i)
 		Gg(fid, func() {
 			out, err := f(ctx)

intra/core/expiringmap.go

@@ -48,7 +48,7 @@ func NewExpiringMapLifetime[P comparable, Q any](ctx context.Context, min time.D
 		lastreap: time.Now(),
 		minlife:  min,
 	}
-	go m.reaper(ctx)
+	Gx1("expm.reaper", m.reaper, ctx)
 	// test: go.dev/play/p/EYq_STKvugb
 	return m
 }

intra/core/p2est.go

@@ -66,7 +66,7 @@ func NewP2QuantileEstimator(ctx context.Context, samples int64, probability floa
 		count: 0,
 		addc:  make(chan float64, samples),
 	}
-	go p.run()
+	Gx("p2.est", p.run)
 	return p
 }
 

intra/core/volatileflow.go

@@ -67,7 +67,7 @@ func NewFlowFor[T any](ctx context.Context, v *Volatile[T]) *Flow[T] {
 		c:   make(chan T),
 		ctx: ctx,
 	}
-	go f.stream()
+	Gx("core.flow", f.stream)
 	return f
 }
 
@@ -78,25 +78,25 @@ func (f *Flow[T]) stream() {
 		case <-f.ctx.Done():
 			return
 		case v := <-f.c:
-			go func() {
+			Gx("flow.stream", func() {
 				notflowing := make(map[FlowOn[T]]struct{}, 0)
 				for _, o := range f.observers() {
 					if ok := o.flow(v); !ok {
 						notflowing[o] = struct{}{}
 					}
 				}
 				f.removeFinallys(notflowing)
-			}()
+			})
 		case <-time.Tick(3 * time.Hour):
-			go func() {
+			Gx("flow.stream.tick", func() {
 				notflowing := make(map[FlowOn[T]]struct{})
 				for _, o := range f.observers() {
 					if o.obsolete() {
 						notflowing[o] = struct{}{}
 					}
 				}
 				f.removeFinallys(notflowing)
-			}()
+			})
 		}
 	}
 }
@@ -151,7 +151,7 @@ func (f *Flow[T]) On(until context.Context, o FlowFunc[T]) {
 	defer f.fmu.Unlock()
 	on := FlowOn[T]{until, &o}
 	f.o = append(f.o, on)
-	go on.flow(f.v.Load())
+	Gx("flow.on", func() { on.flow(f.v.Load()) })
 }
 
 func (f *Flow[T]) Store(v T) {

intra/dns53/mdns.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"net"
 	"net/netip"
+	"runtime/debug"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -548,6 +549,7 @@ func (c *client) recv(conn net.PacketConn) {
 		return
 	}
 
+	debug.SetPanicOnFault(true)
 	defer core.Recover(core.DontExit, "mdns.recv")
 
 	bptr := core.Alloc()

intra/dnscrypt/multiserver.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"net"
 	"net/netip"
+	"runtime/debug"
 	"strings"
 	"sync"
 	"time"
@@ -481,6 +482,7 @@ func (proxy *DcMulti) Stop() error {
 // refreshRoutes re-adds relay routes to all live/tracked servers.
 // Must be called from a goroutine.
 func (proxy *DcMulti) refreshRoutes() {
+	debug.SetPanicOnFault(true)
 	defer core.Recover(core.Exit11, "dcmulti.refreshRoutes")
 
 	udp, tcp := route(proxy)

intra/dnsx/alg.go

@@ -1020,7 +1020,7 @@ func (t *dnsgateway) qs(t2 Transport, uid, network string, msg *dns.Msg, t1res <
 	t2res := make(chan secans, 1)
 	msg = msg.Copy() // to avoid racing against changes made by caller
 
-	go func() {
+	core.Gx("alg.qs."+xdns.QName(msg), func() {
 		defer close(t2res)
 
 		qname := xdns.QName(msg)
@@ -1036,7 +1036,7 @@ func (t *dnsgateway) qs(t2 Transport, uid, network string, msg *dns.Msg, t1res <
 		r.initIfNeeded() // r may be nil value on Grx:timeout
 
 		t2res <- r // may be zero secans
-	}()
+	})
 	return t2res
 }
 

intra/dnsx/cacher.go

@@ -204,7 +204,6 @@ func mkcachekey(q *dns.Msg) (string, uint8, bool) {
 // scrubCache deletes expired entries from the cache.
 // Must be called from a goroutine.
 func (cb *cache) scrubCache() {
-	defer core.Recover(core.Exit11, "c.scrubCache")
 	// must unlock from deferred since panics are recovered above
 	cb.mu.Lock()
 	defer cb.mu.Unlock()
@@ -281,7 +280,7 @@ func (cb *cache) put(key string, cc *cres) (ok bool) {
 	defer cb.mu.Unlock()
 
 	if rand33pc() { // 33% of the time
-		go cb.scrubCache()
+		core.Gx("c.scrubCache", cb.scrubCache)
 	}
 
 	if len(cb.c) >= cb.size {