Browse Source

payments: allocate payment sequences in blocks

The bucket sequence we use as payment sequence makes the DB update in
InitPayment conflict and therefore queue up when many concurrent
payments happen. This change allocates payment sequences in blocks
of 1000 to avoid these conflicts.
master
Andras Banki-Horvath 3 years ago
parent
commit
f33b5a4057
No known key found for this signature in database
GPG Key ID: 80E5375C094198D8
  1. 74
      channeldb/payment_control.go

74
channeldb/payment_control.go

@ -6,11 +6,18 @@ import (
"errors" "errors"
"fmt" "fmt"
"io" "io"
"sync"
"github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
) )
const (
// paymentSeqBlockSize is the block size used when we batch allocate
// payment sequences for future payments.
paymentSeqBlockSize = 1000
)
var ( var (
// ErrAlreadyPaid signals we have already paid this payment hash. // ErrAlreadyPaid signals we have already paid this payment hash.
ErrAlreadyPaid = errors.New("invoice is already paid") ErrAlreadyPaid = errors.New("invoice is already paid")
@ -84,7 +91,10 @@ var (
// PaymentControl implements persistence for payments and payment attempts. // PaymentControl implements persistence for payments and payment attempts.
type PaymentControl struct { type PaymentControl struct {
db *DB paymentSeqMx sync.Mutex
currPaymentSeq uint64
storedPaymentSeq uint64
db *DB
} }
// NewPaymentControl creates a new instance of the PaymentControl. // NewPaymentControl creates a new instance of the PaymentControl.
@ -101,6 +111,14 @@ func NewPaymentControl(db *DB) *PaymentControl {
func (p *PaymentControl) InitPayment(paymentHash lntypes.Hash, func (p *PaymentControl) InitPayment(paymentHash lntypes.Hash,
info *PaymentCreationInfo) error { info *PaymentCreationInfo) error {
// Obtain a new sequence number for this payment. This is used
// to sort the payments in order of creation, and also acts as
// a unique identifier for each payment.
sequenceNum, err := p.nextPaymentSequence()
if err != nil {
return err
}
var b bytes.Buffer var b bytes.Buffer
if err := serializePaymentCreationInfo(&b, info); err != nil { if err := serializePaymentCreationInfo(&b, info); err != nil {
return err return err
@ -108,7 +126,7 @@ func (p *PaymentControl) InitPayment(paymentHash lntypes.Hash,
infoBytes := b.Bytes() infoBytes := b.Bytes()
var updateErr error var updateErr error
err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error { err = kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
// Reset the update error, to avoid carrying over an error // Reset the update error, to avoid carrying over an error
// from a previous execution of the batched db transaction. // from a previous execution of the batched db transaction.
updateErr = nil updateErr = nil
@ -150,14 +168,6 @@ func (p *PaymentControl) InitPayment(paymentHash lntypes.Hash,
return nil return nil
} }
// Obtain a new sequence number for this payment. This is used
// to sort the payments in order of creation, and also acts as
// a unique identifier for each payment.
sequenceNum, err := nextPaymentSequence(tx)
if err != nil {
return err
}
// Before we set our new sequence number, we check whether this // Before we set our new sequence number, we check whether this
// payment has a previously set sequence number and remove its // payment has a previously set sequence number and remove its
// index entry if it exists. This happens in the case where we // index entry if it exists. This happens in the case where we
@ -615,19 +625,45 @@ func fetchPaymentBucketUpdate(tx kvdb.RwTx, paymentHash lntypes.Hash) (
// nextPaymentSequence returns the next sequence number to store for a new // nextPaymentSequence returns the next sequence number to store for a new
// payment. // payment.
func nextPaymentSequence(tx kvdb.RwTx) ([]byte, error) { func (p *PaymentControl) nextPaymentSequence() ([]byte, error) {
payments, err := tx.CreateTopLevelBucket(paymentsRootBucket) p.paymentSeqMx.Lock()
if err != nil { defer p.paymentSeqMx.Unlock()
return nil, err
} // Set a new upper bound in the DB every 1000 payments to avoid
// conflicts on the sequence when using etcd.
if p.currPaymentSeq == p.storedPaymentSeq {
var currPaymentSeq, newUpperBound uint64
if err := kvdb.Update(p.db.Backend, func(tx kvdb.RwTx) error {
paymentsBucket, err := tx.CreateTopLevelBucket(
paymentsRootBucket,
)
if err != nil {
return err
}
seq, err := payments.NextSequence() currPaymentSeq = paymentsBucket.Sequence()
if err != nil { newUpperBound = currPaymentSeq + paymentSeqBlockSize
return nil, err return paymentsBucket.SetSequence(newUpperBound)
}, func() {}); err != nil {
return nil, err
}
// We lazy initialize the cached currPaymentSeq here using the
// first nextPaymentSequence() call. This if statement will auto
// initialize our stored currPaymentSeq, since by default both
// this variable and storedPaymentSeq are zero which in turn
// will have us fetch the current values from the DB.
if p.currPaymentSeq == 0 {
p.currPaymentSeq = currPaymentSeq
}
p.storedPaymentSeq = newUpperBound
} }
p.currPaymentSeq++
b := make([]byte, 8) b := make([]byte, 8)
binary.BigEndian.PutUint64(b, seq) binary.BigEndian.PutUint64(b, p.currPaymentSeq)
return b, nil return b, nil
} }

Loading…
Cancel
Save