Add Kmutex

common/kmutex/mutex.go

@@ -0,0 +1,57 @@
+package kmutex
+
+import "sync"
+
+type Kmutex[T comparable] struct {
+	l sync.Locker
+	s map[T]*klock
+}
+
+type klock struct {
+	cond *sync.Cond
+	ref  uint64
+}
+
+// Create new Kmutex
+func New[T comparable]() *Kmutex[T] {
+	l := sync.Mutex{}
+	return &Kmutex[T]{
+		l: &l,
+		s: make(map[T]*klock),
+	}
+}
+
+// Unlock Kmutex by unique ID
+func (km *Kmutex[T]) Unlock(key T) {
+	km.l.Lock()
+	defer km.l.Unlock()
+	kl, ok := km.s[key]
+	if !ok || kl.ref == 0 {
+		panic("unlock of unlocked kmutex")
+	}
+	kl.ref--
+	if kl.ref == 0 {
+		delete(km.s, key)
+		return
+	}
+	kl.cond.Signal()
+}
+
+// Lock Kmutex by unique ID
+func (km *Kmutex[T]) Lock(key T) {
+	km.l.Lock()
+	defer km.l.Unlock()
+	for {
+		kl, ok := km.s[key]
+		if !ok {
+			km.s[key] = &klock{
+				cond: sync.NewCond(km.l),
+				ref:  1,
+			}
+			return
+		}
+		kl.ref++
+		kl.cond.Wait()
+		return
+	}
+}

common/kmutex/mutex_test.go

@@ -0,0 +1,96 @@
+package kmutex
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+// Number of unique resources to access
+const number = 100
+
+func makeIds(count int) []int {
+	ids := make([]int, count)
+	for i := 0; i < count; i++ {
+		ids[i] = i
+	}
+	return ids
+}
+
+func TestKmutex(t *testing.T) {
+	km := New[int]()
+	ids := makeIds(number)
+	resources := make([]int, number)
+	wg := sync.WaitGroup{}
+
+	lc := make(chan int)
+	uc := make(chan int)
+	// Start 10n goroutines accessing n resources 10 times each
+	for i := 0; i < 10*number; i++ {
+		wg.Add(1)
+		go func(k int) {
+			for j := 0; j < 10; j++ {
+				lc <- k
+				km.Lock(ids[k])
+				// read and write resource to check for race
+				resources[k] = resources[k] + 1
+				km.Unlock(ids[k])
+				uc <- k
+			}
+			wg.Done()
+		}(i % len(ids))
+	}
+
+	to := time.After(time.Second)
+	counts := make(map[int]int)
+	var lCount, ulCount int
+loop:
+	for {
+		select {
+		case k := <-lc:
+			counts[k] = counts[k] + 1
+			lCount++
+		case k := <-uc:
+			counts[k] = counts[k] - 1
+			ulCount++
+		case <-to:
+			t.Fatal("timed out waiting for results")
+			break loop
+		}
+		expectCount := 100 * number
+		if lCount == expectCount && ulCount == expectCount {
+			// Have all results
+			break
+		}
+	}
+	for k, c := range counts {
+		if c != 0 {
+			t.Errorf("Key %d count != 0: %d\n", k, c)
+		}
+	}
+
+	wg.Wait()
+}
+
+func BenchmarkKmutex1000(b *testing.B) {
+	km := New[int]()
+	ids := makeIds(number)
+	resources := make([]int, number)
+	wg := sync.WaitGroup{}
+
+	// Start 1000 goroutines accessing 100 resources N times each
+	b.ResetTimer()
+	for i := 0; i < 1000; i++ {
+		wg.Add(1)
+		go func(k int) {
+			for j := 0; j < b.N; j++ {
+				km.Lock(ids[k])
+				// read and write resource to check for race
+				resources[k] = resources[k] + 1
+				km.Unlock(ids[k])
+			}
+			wg.Done()
+		}(i % len(ids))
+	}
+	wg.Wait()
+}

common/migrate/source/raw.go

@@ -0,0 +1,98 @@
+package source
+
+import (
+	"io"
+	"io/fs"
+	"io/ioutil"
+	"strconv"
+	"strings"
+
+	"github.com/golang-migrate/migrate/v4/source"
+	E "github.com/sagernet/sing/common/exceptions"
+)
+
+type RawDriver struct {
+	migrations    *source.Migrations
+	rawMigrations map[string]string
+}
+
+func NewRawDriver(rawMigrations map[string]string) *RawDriver {
+	return &RawDriver{rawMigrations: rawMigrations}
+}
+
+func (d *RawDriver) Init() error {
+	ms := source.NewMigrations()
+	for key := range d.rawMigrations {
+		m, err := source.DefaultParse(key)
+		if err != nil {
+			continue
+		}
+		if !ms.Append(m) {
+			return source.ErrDuplicateMigration{
+				Migration: *m,
+			}
+		}
+	}
+	d.migrations = ms
+	return nil
+}
+
+func (d *RawDriver) Open(url string) (source.Driver, error) {
+	return nil, E.New("open() cannot be called")
+}
+
+func (d *RawDriver) Close() error {
+	return nil
+}
+
+func (d *RawDriver) First() (version uint, err error) {
+	if version, ok := d.migrations.First(); ok {
+		return version, nil
+	}
+	return 0, &fs.PathError{
+		Op:  "first",
+		Err: fs.ErrNotExist,
+	}
+}
+
+func (d *RawDriver) Prev(version uint) (prevVersion uint, err error) {
+	if version, ok := d.migrations.Prev(version); ok {
+		return version, nil
+	}
+	return 0, &fs.PathError{
+		Op:  "prev for version " + strconv.FormatUint(uint64(version), 10),
+		Err: fs.ErrNotExist,
+	}
+}
+
+func (d *RawDriver) Next(version uint) (nextVersion uint, err error) {
+	if version, ok := d.migrations.Next(version); ok {
+		return version, nil
+	}
+	return 0, &fs.PathError{
+		Op:  "next for version " + strconv.FormatUint(uint64(version), 10),
+		Err: fs.ErrNotExist,
+	}
+}
+
+func (d *RawDriver) ReadUp(version uint) (r io.ReadCloser, identifier string, err error) {
+	if m, ok := d.migrations.Up(version); ok {
+		body := ioutil.NopCloser(strings.NewReader(d.rawMigrations[m.Raw]))
+		return body, m.Identifier, nil
+	}
+	return nil, "", &fs.PathError{
+		Op:  "read up for version " + strconv.FormatUint(uint64(version), 10),
+		Err: fs.ErrNotExist,
+	}
+}
+
+func (d *RawDriver) ReadDown(version uint) (r io.ReadCloser, identifier string, err error) {
+	if m, ok := d.migrations.Down(version); ok {
+		body := ioutil.NopCloser(strings.NewReader(d.rawMigrations[m.Raw]))
+		return body, m.Identifier, nil
+	}
+	return nil, "", &fs.PathError{
+		Op:  "read down for version " + strconv.FormatUint(uint64(version), 10),
+		Err: fs.ErrNotExist,
+	}
+}

common/xray/json/badoption/range.go

@@ -71,6 +71,13 @@ func (c *Range) UnmarshalJSON(content []byte) error {
 	return nil
 }
 
+func (c *Range) String() string {
+	if c.From == c.To {
+		return strconv.FormatInt(int64(c.From), 10)
+	}
+	return fmt.Sprintf("%d-%d", c.From, c.To)
+}
+
 func (c Range) Rand() int32 {
 	return int32(crypto.RandBetween(int64(c.From), int64(c.To)))
 }

constant/proxy.go

@@ -97,18 +97,22 @@ func ProxyDisplayName(proxyType string) string {
 		return "TUIC"
 	case TypeHysteria2:
 		return "Hysteria2"
+	case TypeBond:
+		return "Bond"
 	case TypeMieru:
 		return "Mieru"
 	case TypeAnyTLS:
 		return "AnyTLS"
+	case TypeFailover:
+		return "Failover"
 	case TypeSelector:
 		return "Selector"
 	case TypeURLTest:
 		return "URLTest"
 	case TypeTunnelClient:
-		return "Tunnel Client"
+		return "Tunnel client"
 	case TypeTunnelServer:
-		return "Tunnel Server"
+		return "Tunnel server"
 	default:
 		return "Unknown"
 	}

option/wireguard.go

@@ -3,6 +3,7 @@ package option
 import (
 	"net/netip"
 
+	Xbadoption "github.com/sagernet/sing-box/common/xray/json/badoption"
 	"github.com/sagernet/sing/common/json/badoption"
 )
 
@@ -84,24 +85,24 @@ type LegacyWireGuardPeer struct {
 }
 
 type WireGuardAmnezia struct {
-	JC    int    `json:"jc,omitempty"`
-	JMin  int    `json:"jmin,omitempty"`
-	JMax  int    `json:"jmax,omitempty"`
-	S1    int    `json:"s1,omitempty"`
-	S2    int    `json:"s2,omitempty"`
-	S3    int    `json:"s3,omitempty"`
-	S4    int    `json:"s4,omitempty"`
-	H1    uint32 `json:"h1,omitempty"`
-	H2    uint32 `json:"h2,omitempty"`
-	H3    uint32 `json:"h3,omitempty"`
-	H4    uint32 `json:"h4,omitempty"`
-	I1    string `json:"i1,omitempty"`
-	I2    string `json:"i2,omitempty"`
-	I3    string `json:"i3,omitempty"`
-	I4    string `json:"i4,omitempty"`
-	I5    string `json:"i5,omitempty"`
-	J1    string `json:"j1,omitempty"`
-	J2    string `json:"j2,omitempty"`
-	J3    string `json:"j3,omitempty"`
-	ITime int64  `json:"itime,omitempty"`
+	JC    int               `json:"jc,omitempty"`
+	JMin  int               `json:"jmin,omitempty"`
+	JMax  int               `json:"jmax,omitempty"`
+	S1    int               `json:"s1,omitempty"`
+	S2    int               `json:"s2,omitempty"`
+	S3    int               `json:"s3,omitempty"`
+	S4    int               `json:"s4,omitempty"`
+	H1    *Xbadoption.Range `json:"h1,omitempty"`
+	H2    *Xbadoption.Range `json:"h2,omitempty"`
+	H3    *Xbadoption.Range `json:"h3,omitempty"`
+	H4    *Xbadoption.Range `json:"h4,omitempty"`
+	I1    string            `json:"i1,omitempty"`
+	I2    string            `json:"i2,omitempty"`
+	I3    string            `json:"i3,omitempty"`
+	I4    string            `json:"i4,omitempty"`
+	I5    string            `json:"i5,omitempty"`
+	J1    string            `json:"j1,omitempty"`
+	J2    string            `json:"j2,omitempty"`
+	J3    string            `json:"j3,omitempty"`
+	ITime int64             `json:"itime,omitempty"`
 }

protocol/bond/inbound.go

@@ -4,12 +4,12 @@ import (
 	"context"
 	"errors"
 	"net"
-	"sync"
 	"time"
 
 	"github.com/patrickmn/go-cache"
 	"github.com/sagernet/sing-box/adapter"
 	"github.com/sagernet/sing-box/adapter/inbound"
+	"github.com/sagernet/sing-box/common/kmutex"
 	"github.com/sagernet/sing-box/common/uot"
 	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing-box/log"
@@ -31,7 +31,7 @@ type Inbound struct {
 	inbounds []adapter.Inbound
 	conns    *cache.Cache
 
-	mtx sync.Mutex
+	mtx *kmutex.Kmutex[string]
 }
 
 func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.BondInboundOptions) (adapter.Inbound, error) {
@@ -39,10 +39,11 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
 		return nil, E.New("missing tags")
 	}
 	inbound := &Inbound{
-		Adapter: inbound.NewAdapter(C.TypeTunnelServer, tag),
+		Adapter: inbound.NewAdapter(C.TypeBond, tag),
 		logger:  logger,
 		router:  uot.NewRouter(router, logger),
 		conns:   cache.New(C.TCPConnectTimeout, time.Second),
+		mtx:     kmutex.New[string](),
 	}
 	inboundRegistry := service.FromContext[adapter.InboundRegistry](ctx)
 	inbounds := make([]adapter.Inbound, len(options.Inbounds))
@@ -55,8 +56,8 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
 	}
 	inbound.inbounds = inbounds
 	inbound.conns.OnEvicted(func(s string, i interface{}) {
-		inbound.mtx.Lock()
-		defer inbound.mtx.Unlock()
+		inbound.mtx.Lock(s)
+		defer inbound.mtx.Unlock(s)
 		ratioConns := i.(map[uint8]*ratioConn)
 		for _, ratioConn := range ratioConns {
 			if ratioConn != nil {
@@ -100,15 +101,15 @@ func (h *Inbound) connHandler(ctx context.Context, conn net.Conn, metadata adapt
 	if err != nil {
 		return err
 	}
-	h.mtx.Lock()
-	defer h.mtx.Unlock()
+	requestUUID := request.UUID.String()
+	h.mtx.Lock(requestUUID)
 	var ratioConns map[uint8]*ratioConn
-	rawRatioConns, ok := h.conns.Get(request.UUID.String())
+	rawRatioConns, ok := h.conns.Get(requestUUID)
 	if ok {
 		ratioConns = rawRatioConns.(map[uint8]*ratioConn)
 	} else {
 		ratioConns = make(map[uint8]*ratioConn, request.Count)
-		h.conns.SetDefault(request.UUID.String(), ratioConns)
+		h.conns.SetDefault(requestUUID, ratioConns)
 	}
 	ratioConns[request.Index] = &ratioConn{
 		conn:          conn,
@@ -132,14 +133,18 @@ func (h *Inbound) connHandler(ctx context.Context, conn net.Conn, metadata adapt
 			for _, conn := range conns {
 				conn.Close()
 			}
+			h.mtx.Unlock(requestUUID)
 			return E.New("invalid ratios")
 		}
 		conn = NewBondedConn(conns, downloadRatios, uploadRatios)
 		metadata.Inbound = h.Tag()
 		metadata.InboundType = C.TypeBond
 		metadata.Destination = request.Destination
+		h.mtx.Unlock(requestUUID)
 		h.router.RouteConnectionEx(ctx, conn, metadata, onClose)
+		return nil
 	}
+	h.mtx.Unlock(requestUUID)
 	return nil
 }
 

protocol/bond/outbound.go

@@ -62,7 +62,7 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL
 		return nil, E.New("invalid ratios")
 	}
 	outbound := &Outbound{
-		Adapter:        outbound.NewAdapter(C.TypeTunnelClient, tag, []string{N.NetworkTCP, N.NetworkUDP}, []string{}),
+		Adapter:        outbound.NewAdapter(C.TypeBond, tag, []string{N.NetworkTCP, N.NetworkUDP}, []string{}),
 		ctx:            ctx,
 		outbounds:      outbounds,
 		downloadRatios: downloadRatios,

protocol/group/failover.go

@@ -100,6 +100,9 @@ func (s *Failover) ListenPacket(ctx context.Context, destination M.Socksaddr) (n
 			s.logger.ErrorContext(ctx, err)
 			continue
 		}
+		s.mtx.Lock()
+		defer s.mtx.Unlock()
+		s.lastUsedOutbound = outbound.Tag()
 		return conn, nil
 	}
 	return nil, err

transport/wireguard/endpoint.go

@@ -202,17 +202,17 @@ func (e *Endpoint) Start(resolve bool) error {
 		if e.options.Amnezia.S4 > 0 {
 			ipcConf += "\ns4=" + strconv.Itoa(e.options.Amnezia.S4)
 		}
-		if e.options.Amnezia.H1 > 0 {
-			ipcConf += "\nh1=" + strconv.FormatUint(uint64(e.options.Amnezia.H1), 10)
+		if e.options.Amnezia.H1 != nil {
+			ipcConf += "\nh1=" + e.options.Amnezia.H1.String()
 		}
-		if e.options.Amnezia.H2 > 0 {
-			ipcConf += "\nh2=" + strconv.FormatUint(uint64(e.options.Amnezia.H2), 10)
+		if e.options.Amnezia.H2 != nil {
+			ipcConf += "\nh2=" + e.options.Amnezia.H2.String()
 		}
-		if e.options.Amnezia.H3 > 0 {
-			ipcConf += "\nh3=" + strconv.FormatUint(uint64(e.options.Amnezia.H3), 10)
+		if e.options.Amnezia.H3 != nil {
+			ipcConf += "\nh3=" + e.options.Amnezia.H3.String()
 		}
-		if e.options.Amnezia.H4 > 0 {
-			ipcConf += "\nh4=" + strconv.FormatUint(uint64(e.options.Amnezia.H4), 10)
+		if e.options.Amnezia.H4 != nil {
+			ipcConf += "\nh4=" + e.options.Amnezia.H4.String()
 		}
 		if e.options.Amnezia.I1 != "" {
 			ipcConf += "\ni1=" + e.options.Amnezia.I1

transport/wireguard/endpoint_options.go

@@ -5,6 +5,7 @@ import (
 	"net/netip"
 	"time"
 
+	Xbadoption "github.com/sagernet/sing-box/common/xray/json/badoption"
 	tun "github.com/sagernet/sing-tun"
 	"github.com/sagernet/sing/common/logger"
 	M "github.com/sagernet/sing/common/metadata"
@@ -49,10 +50,10 @@ type AmneziaOptions struct {
 	S2    int
 	S3    int
 	S4    int
-	H1    uint32
-	H2    uint32
-	H3    uint32
-	H4    uint32
+	H1    *Xbadoption.Range
+	H2    *Xbadoption.Range
+	H3    *Xbadoption.Range
+	H4    *Xbadoption.Range
 	I1    string
 	I2    string
 	I3    string