les: add atomicWriteLock to client db

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
+	posBalancePrefix      = []byte("pb:")         // dbVersion(uint16 big endian) + posBalancePrefix + id -> positive balance
-	negBalancePrefix = []byte("nb:")         // dbVersion(uint16 big endian) + negBalancePrefix + ip -> negative balance
+	negBalancePrefix      = []byte("nb:")         // dbVersion(uint16 big endian) + negBalancePrefix + ip -> negative balance
-	curBalancePrefix = []byte("cb:")         // dbVersion(uint16 big endian) + curBalancePrefix + id -> currency balance
+	curBalancePrefix      = []byte("cb:")         // dbVersion(uint16 big endian) + curBalancePrefix + id -> currency balance
-	expirationKey    = []byte("expiration:") // dbVersion(uint16 big endian) + expirationKey -> posExp, negExp
+	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)
+	key := idKey(id, neg)
-	db.db.Delete(key)
+	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() {

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)
+	receivePayment(from enode.ID, proofOfPayment []byte, reader ethdb.KeyValueReader, writer ethdb.KeyValueWriter) (value uint64, err error)
-	requestPayment(from enode.ID, value uint64) uint64
+	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...))
+}