diff --git a/les/clientdb.go b/les/clientdb.go index b1beeac915..f24d499479 100644 --- a/les/clientdb.go +++ b/les/clientdb.go @@ -20,6 +20,7 @@ import ( "bytes" "encoding/binary" "io" + "sync" "time" "github.com/ethereum/go-ethereum/common" @@ -95,19 +96,27 @@ const ( ) var ( - posBalancePrefix = []byte("pb:") // dbVersion(uint16 big endian) + posBalancePrefix + id -> positive balance - negBalancePrefix = []byte("nb:") // dbVersion(uint16 big endian) + negBalancePrefix + ip -> negative balance - curBalancePrefix = []byte("cb:") // dbVersion(uint16 big endian) + curBalancePrefix + id -> currency balance - expirationKey = []byte("expiration:") // dbVersion(uint16 big endian) + expirationKey -> posExp, negExp + posBalancePrefix = []byte("pb:") // dbVersion(uint16 big endian) + posBalancePrefix + id -> positive balance + negBalancePrefix = []byte("nb:") // dbVersion(uint16 big endian) + negBalancePrefix + ip -> negative balance + curBalancePrefix = []byte("cb:") // dbVersion(uint16 big endian) + curBalancePrefix + id -> currency balance + paymentReceiverPrefix = []byte("pr:") // dbVersion(uint16 big endian) + paymentReceiverPrefix + id + "receiverName:" -> receiver namespace + expirationKey = []byte("expiration:") // dbVersion(uint16 big endian) + expirationKey -> posExp, negExp ) +type atomicWriteLock struct { + released chan struct{} + batch ethdb.Batch +} + type nodeDB struct { db ethdb.Database cache *lru.Cache clock mclock.Clock closeCh chan struct{} evictCallBack func(mclock.AbsTime, bool, tokenBalance) bool // Callback to determine whether the balance can be evicted. - cleanupHook func() // Test hook used for testing + idLockMutex sync.Mutex + idLocks map[string]atomicWriteLock + cleanupHook func() // Test hook used for testing } func newNodeDB(db ethdb.Database, clock mclock.Clock) *nodeDB { @@ -120,6 +129,7 @@ func newNodeDB(db ethdb.Database, clock mclock.Clock) *nodeDB { cache: cache, clock: clock, closeCh: make(chan struct{}), + idLocks: make(map[string]atomicWriteLock), } go ndb.expirer() return ndb @@ -129,7 +139,46 @@ func (db *nodeDB) close() { close(db.closeCh) } -func (db *nodeDB) key(id []byte, neg bool) []byte { +func (db *nodeDB) atomicWriteLock(id []byte) ethdb.KeyValueWriter { + db.idLockMutex.Lock() + for { + ch := db.idLocks[string(id)].released + if ch == nil { + break + } + db.idLockMutex.Unlock() + <-ch + db.idLockMutex.Lock() + } + batch := db.db.NewBatch() + db.idLocks[string(id)] = atomicWriteLock{ + released: make(chan struct{}), + batch: batch, + } + db.idLockMutex.Unlock() + return batch +} + +func (db *nodeDB) atomicWriteUnlock(id []byte) { + db.idLockMutex.Lock() + awl := db.idLocks[string(id)] + awl.batch.Write() + close(awl.released) + delete(db.idLocks, string(id)) + db.idLockMutex.Unlock() +} + +func (db *nodeDB) writer(id []byte) ethdb.KeyValueWriter { + db.idLockMutex.Lock() + batch := db.idLocks[string(id)].batch + db.idLockMutex.Unlock() + if batch == nil { + return db.db + } + return batch +} + +func idKey(id []byte, neg bool) []byte { prefix := posBalancePrefix if neg { prefix = negBalancePrefix @@ -137,6 +186,10 @@ func (db *nodeDB) key(id []byte, neg bool) []byte { return append(prefix, id...) } +func receiverPrefix(id enode.ID, receiver string) []byte { + return append(append(paymentReceiverPrefix, id.Bytes()...), []byte(receiver+":")...) +} + func (db *nodeDB) getExpiration() (fixed64, fixed64) { blob, err := db.db.Get(expirationKey) if err != nil || len(blob) != 16 { @@ -169,11 +222,11 @@ func (db *nodeDB) setCurrencyBalance(id enode.ID, b currencyBalance) { if err != nil { log.Crit("Failed to encode currency balance", "err", err) } - db.db.Put(append(curBalancePrefix, id.Bytes()...), enc) + db.writer(id.Bytes()).Put(append(curBalancePrefix, id.Bytes()...), enc) } func (db *nodeDB) getOrNewBalance(id []byte, neg bool) tokenBalance { - key := db.key(id, neg) + key := idKey(id, neg) item, exist := db.cache.Get(string(key)) if exist { return item.(tokenBalance) @@ -191,18 +244,26 @@ func (db *nodeDB) getOrNewBalance(id []byte, neg bool) tokenBalance { } func (db *nodeDB) setBalance(id []byte, neg bool, b tokenBalance) { - key := db.key(id, neg) + key := idKey(id, neg) enc, err := rlp.EncodeToBytes(&(b)) if err != nil { log.Crit("Failed to encode positive balance", "err", err) } - db.db.Put(key, enc) + if neg { + db.db.Put(key, enc) + } else { + db.writer(id).Put(key, enc) + } db.cache.Add(string(key), b) } func (db *nodeDB) delBalance(id []byte, neg bool) { - key := db.key(id, neg) - db.db.Delete(key) + key := idKey(id, neg) + if neg { + db.db.Delete(key) + } else { + db.writer(id).Delete(key) + } db.cache.Remove(string(key)) } @@ -212,7 +273,7 @@ func (db *nodeDB) getPosBalanceIDs(start, stop enode.ID, maxCount int) (result [ if maxCount <= 0 { return } - it := db.db.NewIteratorWithStart(db.key(start.Bytes(), false)) + it := db.db.NewIteratorWithStart(idKey(start.Bytes(), false)) defer it.Release() for i := len(stop[:]) - 1; i >= 0; i-- { stop[i]-- @@ -220,7 +281,7 @@ func (db *nodeDB) getPosBalanceIDs(start, stop enode.ID, maxCount int) (result [ break } } - stopKey := db.key(stop.Bytes(), false) + stopKey := idKey(stop.Bytes(), false) keyLen := len(stopKey) for it.Next() { diff --git a/les/tokensale.go b/les/tokensale.go index b00fcfa776..099058fa71 100644 --- a/les/tokensale.go +++ b/les/tokensale.go @@ -25,6 +25,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/rlp" ) @@ -40,8 +41,8 @@ const ( // payment technologies type paymentReceiver interface { info() keyValueList - receivePayment(from enode.ID, proofOfPayment []byte) (value uint64, err error) - requestPayment(from enode.ID, value uint64) uint64 + receivePayment(from enode.ID, proofOfPayment []byte, reader ethdb.KeyValueReader, writer ethdb.KeyValueWriter) (value uint64, err error) + requestPayment(from enode.ID, value uint64, reader ethdb.KeyValueReader) uint64 } // tokenSale handles client balance deposits, conversion to and from service tokens @@ -437,7 +438,7 @@ func (t *tokenSale) connection(id enode.ID, freeID string, requestedCapacity uin if rec, ok := t.receivers[recID]; !ok || pcMissing == math.MaxUint64 { paymentRequired[i] = math.MaxUint64 } else { - paymentRequired[i] = rec.requestPayment(id, pcMissing) + paymentRequired[i] = rec.requestPayment(id, pcMissing, newReaderTable(t.clientPool.ndb.db, receiverPrefix(id, recID))) } } return @@ -445,8 +446,12 @@ func (t *tokenSale) connection(id enode.ID, freeID string, requestedCapacity uin // deposit credits a payment on the sender's account using the specified payment module func (t *tokenSale) deposit(id enode.ID, paymentModule string, proofOfPayment []byte) (pcValue, pcBalance uint64, err error) { + writer := t.clientPool.ndb.atomicWriteLock(id.Bytes()) t.lock.Lock() - defer t.lock.Unlock() + defer func() { + t.lock.Unlock() + t.clientPool.ndb.atomicWriteUnlock(id.Bytes()) + }() cb := t.clientPool.ndb.getCurrencyBalance(id) pcBalance = cb.amount @@ -454,7 +459,8 @@ func (t *tokenSale) deposit(id enode.ID, paymentModule string, proofOfPayment [] if pm == nil { return 0, pcBalance, fmt.Errorf("Unknown payment receiver '%s'", paymentModule) } - pcValue, err = pm.receivePayment(id, proofOfPayment) + prefix := receiverPrefix(id, paymentModule) + pcValue, err = pm.receivePayment(id, proofOfPayment, newReaderTable(t.clientPool.ndb.db, prefix), newWriterTable(writer, prefix)) if err != nil { return 0, pcBalance, err } @@ -478,8 +484,12 @@ func (t *tokenSale) deposit(id enode.ID, paymentModule string, proofOfPayment [] // is impossible to do a conversion twice. In exchange the sender needs to know its current // balances (which it probably does if it has made a previous call to just ask the current price). func (t *tokenSale) buyTokens(id enode.ID, maxSpend, minReceive uint64, relative, spendAll bool) (pcBalance, tokenBalance, spend, receive uint64, success bool) { + t.clientPool.ndb.atomicWriteLock(id.Bytes()) t.lock.Lock() - defer t.lock.Unlock() + defer func() { + t.lock.Unlock() + t.clientPool.ndb.atomicWriteUnlock(id.Bytes()) + }() pb := t.clientPool.getPosBalance(id) tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now())) @@ -527,8 +537,12 @@ func (t *tokenSale) buyTokens(id enode.ID, maxSpend, minReceive uint64, relative // sellTokens tries to convert service tokens to permanent balance (nominated in the server's // preferred currency, PC). Parameters work similarly to buyTokens. func (t *tokenSale) sellTokens(id enode.ID, maxSell, minRefund uint64, relative, sellAll bool) (pcBalance, tokenBalance, sell, refund uint64, success bool) { + t.clientPool.ndb.atomicWriteLock(id.Bytes()) t.lock.Lock() - defer t.lock.Unlock() + defer func() { + t.lock.Unlock() + t.clientPool.ndb.atomicWriteUnlock(id.Bytes()) + }() pb := t.clientPool.getPosBalance(id) tokenBalance = pb.value.value(t.clientPool.posExpiration(mclock.Now())) @@ -778,7 +792,7 @@ func (t testReceiver) info() keyValueList { // receivePayment implements paymentReceiver. proofOfPayment is a base 10 ascii number // which is credited to the sender's account without any further conditions. -func (t testReceiver) receivePayment(from enode.ID, proofOfPayment []byte) (value uint64, err error) { +func (t testReceiver) receivePayment(from enode.ID, proofOfPayment []byte, reader ethdb.KeyValueReader, writer ethdb.KeyValueWriter) (value uint64, err error) { if len(proofOfPayment) > 8 { err = fmt.Errorf("proof of payment is too long; max 8 bytes long big endian integer expected") return @@ -790,6 +804,57 @@ func (t testReceiver) receivePayment(from enode.ID, proofOfPayment []byte) (valu } // requestPayment implements paymentReceiver -func (t testReceiver) requestPayment(from enode.ID, value uint64) uint64 { +func (t testReceiver) requestPayment(from enode.ID, value uint64, reader ethdb.KeyValueReader) uint64 { return value } + +// readerTable is a wrapper around a database that prefixes each key access with a pre- +// configured string. +type readerTable struct { + db ethdb.KeyValueReader + prefix []byte +} + +// newReaderTable returns a database object that prefixes all keys with a given string. +func newReaderTable(db ethdb.KeyValueReader, prefix []byte) ethdb.KeyValueReader { + return &readerTable{ + db: db, + prefix: prefix, + } +} + +// Has retrieves if a prefixed version of a key is present in the database. +func (t *readerTable) Has(key []byte) (bool, error) { + return t.db.Has(append(t.prefix, key...)) +} + +// Get retrieves the given prefixed key if it's present in the database. +func (t *readerTable) Get(key []byte) ([]byte, error) { + return t.db.Get(append(t.prefix, key...)) +} + +// writerTable is a wrapper around a database that prefixes each key access with a pre- +// configured string. +type writerTable struct { + db ethdb.KeyValueWriter + prefix []byte +} + +// newReaderTable returns a database object that prefixes all keys with a given string. +func newWriterTable(db ethdb.KeyValueWriter, prefix []byte) ethdb.KeyValueWriter { + return &writerTable{ + db: db, + prefix: prefix, + } +} + +// Put inserts the given value into the database at a prefixed version of the +// provided key. +func (t *writerTable) Put(key []byte, value []byte) error { + return t.db.Put(append(t.prefix, key...), value) +} + +// Delete removes the given prefixed key from the database. +func (t writerTable) Delete(key []byte) error { + return t.db.Delete(append(t.prefix, key...)) +}