Introduce mixsplitlimit config option

This allows changing the previously-hardcoded value of 10 allowed CSPP
client connections per any change downmixing amount.

config.go

@@ -55,6 +55,7 @@ const (
 	defaultAccountGapLimit         = wallet.DefaultAccountGapLimit
 	defaultDisableCoinTypeUpgrades = false
 	defaultCircuitLimit            = 32
+	defaultMixSplitLimit           = 10
 
 	// ticket buyer options
 	defaultBalanceToMaintainAbsolute = 0
@@ -164,6 +165,7 @@ type config struct {
 	TicketSplitAccount string `long:"ticketsplitaccount" description:"Account to derive fresh addresses from for mixed ticket splits; uses mixedaccount if unset"`
 	ChangeAccount      string `long:"changeaccount" description:"Account used to derive unmixed CoinJoin outputs in CoinShuffle++ protocol"`
 	MixChange          bool   `long:"mixchange" description:"Use CoinShuffle++ to mix change account outputs into mix account"`
+	MixSplitLimit      int    `long:"mixsplitlimit" description:"Connection limit to CoinShuffle++ server per change amount"`
 
 	TBOpts ticketBuyerOptions `group:"Ticket Buyer Options" namespace:"ticketbuyer"`
 
@@ -363,6 +365,7 @@ func loadConfig(ctx context.Context) (*config, []string, error) {
 		AccountGapLimit:         defaultAccountGapLimit,
 		DisableCoinTypeUpgrades: defaultDisableCoinTypeUpgrades,
 		CircuitLimit:            defaultCircuitLimit,
+		MixSplitLimit:           defaultMixSplitLimit,
 
 		// Ticket Buyer Options
 		TBOpts: ticketBuyerOptions{

dcrwallet.go

@@ -164,7 +164,8 @@ func run(ctx context.Context) error {
 	}
 	loader := ldr.NewLoader(activeNet.Params, dbDir, stakeOptions,
 		cfg.GapLimit, cfg.AllowHighFees, cfg.RelayFee.Amount,
-		cfg.AccountGapLimit, cfg.DisableCoinTypeUpgrades, cfg.ManualTickets)
+		cfg.AccountGapLimit, cfg.DisableCoinTypeUpgrades, cfg.ManualTickets,
+		cfg.MixSplitLimit)
 	loader.DialCSPPServer = cfg.dialCSPPServer
 
 	// Stop any services started by the loader after the shutdown procedure is

internal/loader/loader.go

@@ -46,6 +46,7 @@ type Loader struct {
 	allowHighFees           bool
 	manualTickets           bool
 	relayFee                dcrutil.Amount
+	mixSplitLimit           int
 
 	mu sync.Mutex
 
@@ -69,7 +70,7 @@ type DialFunc func(ctx context.Context, network, addr string) (net.Conn, error)
 
 // NewLoader constructs a Loader.
 func NewLoader(chainParams *chaincfg.Params, dbDirPath string, stakeOptions *StakeOptions, gapLimit uint32,
-	allowHighFees bool, relayFee dcrutil.Amount, accountGapLimit int, disableCoinTypeUpgrades bool, manualTickets bool) *Loader {
+	allowHighFees bool, relayFee dcrutil.Amount, accountGapLimit int, disableCoinTypeUpgrades bool, manualTickets bool, mixSplitLimit int) *Loader {
 
 	return &Loader{
 		chainParams:             chainParams,
@@ -81,6 +82,7 @@ func NewLoader(chainParams *chaincfg.Params, dbDirPath string, stakeOptions *Sta
 		allowHighFees:           allowHighFees,
 		manualTickets:           manualTickets,
 		relayFee:                relayFee,
+		mixSplitLimit:           mixSplitLimit,
 	}
 }
 
@@ -190,6 +192,7 @@ func (l *Loader) CreateWatchingOnlyWallet(ctx context.Context, extendedPubKey st
 		ManualTickets:           l.manualTickets,
 		AllowHighFees:           l.allowHighFees,
 		RelayFee:                l.relayFee,
+		MixSplitLimit:           l.mixSplitLimit,
 		Params:                  l.chainParams,
 	}
 	w, err = wallet.Open(ctx, cfg)
@@ -342,6 +345,7 @@ func (l *Loader) OpenExistingWallet(ctx context.Context, pubPassphrase []byte) (
 		ManualTickets:           l.manualTickets,
 		AllowHighFees:           l.allowHighFees,
 		RelayFee:                l.relayFee,
+		MixSplitLimit:           l.mixSplitLimit,
 		Params:                  l.chainParams,
 	}
 	w, err = wallet.Open(ctx, cfg)

wallet/mixing.go

@@ -39,12 +39,16 @@ var splitPoints = [...]dcrutil.Amount{
 	1 << 18, // 000.00262144
 }
 
-var splitSems = [len(splitPoints)]chan struct{}{}
+type mixSemaphores struct {
+	splitSems [len(splitPoints)]chan struct{}
+}
 
-func init() {
-	for i := range splitSems {
-		splitSems[i] = make(chan struct{}, 10)
+func newMixSemaphores(n int) mixSemaphores {
+	var m mixSemaphores
+	for i := range m.splitSems {
+		m.splitSems[i] = make(chan struct{}, n)
 	}
+	return m
 }
 
 var (
@@ -169,8 +173,8 @@ SplitPoints:
 	select {
 	case <-ctx.Done():
 		return errors.E(op, ctx.Err())
-	case splitSems[i] <- struct{}{}:
-		defer func() { <-splitSems[i] }()
+	case w.mixSems.splitSems[i] <- struct{}{}:
+		defer func() { <-w.mixSems.splitSems[i] }()
 	default:
 		return errThrottledMixRequest
 	}

wallet/wallet.go

@@ -156,6 +156,9 @@ type Wallet struct {
 	passphraseTimeoutMu     sync.Mutex
 	passphraseTimeoutCancel chan struct{}
 
+	// Mix rate limiting
+	mixSems mixSemaphores
+
 	NtfnServer *NotificationServer
 
 	chainParams *chaincfg.Params
@@ -175,6 +178,7 @@ type Config struct {
 
 	GapLimit                uint32
 	AccountGapLimit         int
+	MixSplitLimit           int
 	DisableCoinTypeUpgrades bool
 
 	StakePoolColdExtKey string
@@ -5238,6 +5242,8 @@ func Open(ctx context.Context, cfg *Config) (*Wallet, error) {
 		recentlyPublished: make(map[chainhash.Hash]struct{}),
 
 		addressBuffers: make(map[uint32]*bip0044AccountData),
+
+		mixSems: newMixSemaphores(cfg.MixSplitLimit),
 	}
 
 	// Open database managers

walletsetup.go

@@ -53,7 +53,8 @@ func createWallet(ctx context.Context, cfg *config) error {
 	}
 	loader := loader.NewLoader(activeNet.Params, dbDir, stakeOptions,
 		cfg.GapLimit, cfg.AllowHighFees, cfg.RelayFee.Amount,
-		cfg.AccountGapLimit, cfg.DisableCoinTypeUpgrades, cfg.ManualTickets)
+		cfg.AccountGapLimit, cfg.DisableCoinTypeUpgrades, cfg.ManualTickets,
+		cfg.MixSplitLimit)
 
 	var privPass, pubPass, seed []byte
 	var imported bool