mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-19 02:12:23 +00:00
fixes duplicate naming errors
This commit is contained in:
Eli
parent
527c74902f
commit
36b9870fe6
4 changed files with 84 additions and 84 deletions
|
|
@ -31,7 +31,7 @@ import (
|
|||
var testDownloadDir, _ = ioutil.TempDir(os.TempDir(), "bzz-test")
|
||||
|
||||
func testFileSystem(t *testing.T, f func(*FileSystem)) {
|
||||
testApi(t, func(api *Api) {
|
||||
testAPI(t, func(api *Api) {
|
||||
f(NewFileSystem(api))
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ import (
|
|||
)
|
||||
|
||||
func testStorage(t *testing.T, f func(*Storage)) {
|
||||
testApi(t, func(api *Api) {
|
||||
testAPI(t, func(api *Api) {
|
||||
f(NewStorage(api))
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -101,7 +101,7 @@ Once db backlog is delivered, it reverts back to in-memory buffer
|
|||
It is automatically started when syncdb is initialised.
|
||||
It saves the buffer to db upon receiving quit signal. syncDb#stop()
|
||||
*/
|
||||
func (sdb *syncDb) bufferRead(deliver func(interface{}, chan bool) bool) {
|
||||
func (self *syncDb) bufferRead(deliver func(interface{}, chan bool) bool) {
|
||||
var buffer, db chan interface{} // channels representing the two read modes
|
||||
var more bool
|
||||
var req interface{}
|
||||
|
|
@ -109,18 +109,18 @@ func (sdb *syncDb) bufferRead(deliver func(interface{}, chan bool) bool) {
|
|||
var inBatch, inDb int
|
||||
batch := new(leveldb.Batch)
|
||||
var dbSize chan int
|
||||
quit := sdb.quit
|
||||
quit := self.quit
|
||||
counterValue := make([]byte, 8)
|
||||
|
||||
// counter is used for keeping the items in order, persisted to db
|
||||
// start counter where db was at, 0 if not found
|
||||
data, err := sdb.db.Get(sdb.counterKey)
|
||||
data, err := self.db.Get(self.counterKey)
|
||||
var counter uint64
|
||||
if err == nil {
|
||||
counter = binary.BigEndian.Uint64(data)
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter read from db at %v", sdb.key.Log(), sdb.priority, counter))
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter read from db at %v", self.key.Log(), self.priority, counter))
|
||||
} else {
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter starts at %v", sdb.key.Log(), sdb.priority, counter))
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter starts at %v", self.key.Log(), self.priority, counter))
|
||||
}
|
||||
|
||||
LOOP:
|
||||
|
|
@ -132,26 +132,26 @@ LOOP:
|
|||
// deliver request : this is blocking on network write so
|
||||
// it is passed the quit channel as argument, so that it returns
|
||||
// if syncdb is stopped. In this case we need to save the item to the db
|
||||
more = deliver(req, sdb.quit)
|
||||
more = deliver(req, self.quit)
|
||||
if !more {
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] quit: switching to db. session tally (db/total): %v/%v", sdb.key.Log(), sdb.priority, sdb.dbTotal, sdb.total))
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] quit: switching to db. session tally (db/total): %v/%v", self.key.Log(), self.priority, self.dbTotal, self.total))
|
||||
// received quit signal, save request currently waiting delivery
|
||||
// by switching to db mode and closing the buffer
|
||||
buffer = nil
|
||||
db = sdb.buffer
|
||||
db = self.buffer
|
||||
close(db)
|
||||
quit = nil // needs to block the quit case in select
|
||||
break // break from select, this item will be written to the db
|
||||
}
|
||||
sdb.total++
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] deliver (db/total): %v/%v", sdb.key.Log(), sdb.priority, sdb.dbTotal, sdb.total))
|
||||
self.total++
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] deliver (db/total): %v/%v", self.key.Log(), self.priority, self.dbTotal, self.total))
|
||||
// by the time deliver returns, there were new writes to the buffer
|
||||
// if buffer contention is detected, switch to db mode which drains
|
||||
// the buffer so no process will block on pushing store requests
|
||||
if len(buffer) == cap(buffer) {
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] buffer full %v: switching to db. session tally (db/total): %v/%v", sdb.key.Log(), sdb.priority, cap(buffer), sdb.dbTotal, sdb.total))
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] buffer full %v: switching to db. session tally (db/total): %v/%v", self.key.Log(), self.priority, cap(buffer), self.dbTotal, self.total))
|
||||
buffer = nil
|
||||
db = sdb.buffer
|
||||
db = self.buffer
|
||||
}
|
||||
continue LOOP
|
||||
|
||||
|
|
@ -160,30 +160,30 @@ LOOP:
|
|||
if !more {
|
||||
// only if quit is called, saved all the buffer
|
||||
binary.BigEndian.PutUint64(counterValue, counter)
|
||||
batch.Put(sdb.counterKey, counterValue) // persist counter in batch
|
||||
sdb.writeSyncBatch(batch) // save batch
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save current batch to db", sdb.key.Log(), sdb.priority))
|
||||
batch.Put(self.counterKey, counterValue) // persist counter in batch
|
||||
self.writeSyncBatch(batch) // save batch
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save current batch to db", self.key.Log(), self.priority))
|
||||
break LOOP
|
||||
}
|
||||
sdb.dbTotal++
|
||||
sdb.total++
|
||||
self.dbTotal++
|
||||
self.total++
|
||||
// otherwise break after select
|
||||
case dbSize = <-sdb.batch:
|
||||
case dbSize = <-self.batch:
|
||||
// explicit request for batch
|
||||
if inBatch == 0 && quit != nil {
|
||||
// there was no writes since the last batch so db depleted
|
||||
// switch to buffer mode
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] empty db: switching to buffer", sdb.key.Log(), sdb.priority))
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] empty db: switching to buffer", self.key.Log(), self.priority))
|
||||
db = nil
|
||||
buffer = sdb.buffer
|
||||
buffer = self.buffer
|
||||
dbSize <- 0 // indicates to 'caller' that batch has been written
|
||||
inDb = 0
|
||||
continue LOOP
|
||||
}
|
||||
binary.BigEndian.PutUint64(counterValue, counter)
|
||||
batch.Put(sdb.counterKey, counterValue)
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] write batch %v/%v - %x - %x", sdb.key.Log(), sdb.priority, inBatch, counter, sdb.counterKey, counterValue))
|
||||
batch = sdb.writeSyncBatch(batch)
|
||||
batch.Put(self.counterKey, counterValue)
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] write batch %v/%v - %x - %x", self.key.Log(), self.priority, inBatch, counter, self.counterKey, counterValue))
|
||||
batch = self.writeSyncBatch(batch)
|
||||
dbSize <- inBatch // indicates to 'caller' that batch has been written
|
||||
inBatch = 0
|
||||
continue LOOP
|
||||
|
|
@ -191,45 +191,45 @@ LOOP:
|
|||
// closing syncDb#quit channel is used to signal to all goroutines to quit
|
||||
case <-quit:
|
||||
// need to save backlog, so switch to db mode
|
||||
db = sdb.buffer
|
||||
db = self.buffer
|
||||
buffer = nil
|
||||
quit = nil
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save buffer to db", sdb.key.Log(), sdb.priority))
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save buffer to db", self.key.Log(), self.priority))
|
||||
close(db)
|
||||
continue LOOP
|
||||
}
|
||||
|
||||
// only get here if we put req into db
|
||||
entry, err = sdb.newSyncDbEntry(req, counter)
|
||||
entry, err = self.newSyncDbEntry(req, counter)
|
||||
if err != nil {
|
||||
log.Warn(fmt.Sprintf("syncDb[%v/%v] saving request %v (#%v/%v) failed: %v", sdb.key.Log(), sdb.priority, req, inBatch, inDb, err))
|
||||
log.Warn(fmt.Sprintf("syncDb[%v/%v] saving request %v (#%v/%v) failed: %v", self.key.Log(), self.priority, req, inBatch, inDb, err))
|
||||
continue LOOP
|
||||
}
|
||||
batch.Put(entry.key, entry.val)
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] to batch %v '%v' (#%v/%v/%v)", sdb.key.Log(), sdb.priority, req, entry, inBatch, inDb, counter))
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] to batch %v '%v' (#%v/%v/%v)", self.key.Log(), self.priority, req, entry, inBatch, inDb, counter))
|
||||
// if just switched to db mode and not quitting, then launch dbRead
|
||||
// in a parallel go routine to send deliveries from db
|
||||
if inDb == 0 && quit != nil {
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] start dbRead", sdb.key.Log(), sdb.priority))
|
||||
go sdb.dbRead(true, counter, deliver)
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] start dbRead", self.key.Log(), self.priority))
|
||||
go self.dbRead(true, counter, deliver)
|
||||
}
|
||||
inDb++
|
||||
inBatch++
|
||||
counter++
|
||||
// need to save the batch if it gets too large (== dbBatchSize)
|
||||
if inBatch%int(sdb.dbBatchSize) == 0 {
|
||||
batch = sdb.writeSyncBatch(batch)
|
||||
if inBatch%int(self.dbBatchSize) == 0 {
|
||||
batch = self.writeSyncBatch(batch)
|
||||
}
|
||||
}
|
||||
log.Info(fmt.Sprintf("syncDb[%v:%v]: saved %v keys (saved counter at %v)", sdb.key.Log(), sdb.priority, inBatch, counter))
|
||||
close(sdb.done)
|
||||
log.Info(fmt.Sprintf("syncDb[%v:%v]: saved %v keys (saved counter at %v)", self.key.Log(), self.priority, inBatch, counter))
|
||||
close(self.done)
|
||||
}
|
||||
|
||||
// writes the batch to the db and returns a new batch object
|
||||
func (sdb *syncDb) writeSyncBatch(batch *leveldb.Batch) *leveldb.Batch {
|
||||
err := sdb.db.Write(batch)
|
||||
func (self *syncDb) writeSyncBatch(batch *leveldb.Batch) *leveldb.Batch {
|
||||
err := self.db.Write(batch)
|
||||
if err != nil {
|
||||
log.Warn(fmt.Sprintf("syncDb[%v/%v] saving batch to db failed: %v", sdb.key.Log(), sdb.priority, err))
|
||||
log.Warn(fmt.Sprintf("syncDb[%v/%v] saving batch to db failed: %v", self.key.Log(), self.priority, err))
|
||||
return batch
|
||||
}
|
||||
return new(leveldb.Batch)
|
||||
|
|
@ -240,8 +240,8 @@ type syncDbEntry struct {
|
|||
key, val []byte
|
||||
}
|
||||
|
||||
func (entry syncDbEntry) String() string {
|
||||
return fmt.Sprintf("key: %x, value: %x", entry.key, entry.val)
|
||||
func (self syncDbEntry) String() string {
|
||||
return fmt.Sprintf("key: %x, value: %x", self.key, self.val)
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
@ -262,9 +262,9 @@ dbRead needs a boolean to indicate if on first round all the historical
|
|||
record is synced. Second argument to indicate current db counter
|
||||
The third is the function to apply
|
||||
*/
|
||||
func (sdb *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{}, chan bool) bool) {
|
||||
func (self *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{}, chan bool) bool) {
|
||||
key := make([]byte, 42)
|
||||
copy(key, sdb.start)
|
||||
copy(key, self.start)
|
||||
binary.BigEndian.PutUint64(key[34:], counter)
|
||||
var batches, n, cnt, total int
|
||||
var more bool
|
||||
|
|
@ -280,8 +280,8 @@ func (sdb *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{},
|
|||
// so that loop is not blocking while delivering
|
||||
// only relevant if cnt is large
|
||||
select {
|
||||
case sdb.batch <- batchSizes:
|
||||
case <-sdb.quit:
|
||||
case self.batch <- batchSizes:
|
||||
case <-self.quit:
|
||||
return
|
||||
}
|
||||
// wait for the write to finish and get the item count in the next batch
|
||||
|
|
@ -292,31 +292,31 @@ func (sdb *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{},
|
|||
return
|
||||
}
|
||||
}
|
||||
it = sdb.db.NewIterator()
|
||||
it = self.db.NewIterator()
|
||||
it.Seek(key)
|
||||
if !it.Valid() {
|
||||
copy(key, sdb.start)
|
||||
copy(key, self.start)
|
||||
useBatches = true
|
||||
continue
|
||||
}
|
||||
del = new(leveldb.Batch)
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v]: new iterator: %x (batch %v, count %v)", sdb.key.Log(), sdb.priority, key, batches, cnt))
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v]: new iterator: %x (batch %v, count %v)", self.key.Log(), self.priority, key, batches, cnt))
|
||||
|
||||
for n = 0; !useBatches || n < cnt; it.Next() {
|
||||
copy(key, it.Key())
|
||||
if len(key) == 0 || key[0] != 0 {
|
||||
copy(key, sdb.start)
|
||||
copy(key, self.start)
|
||||
useBatches = true
|
||||
break
|
||||
}
|
||||
val := make([]byte, 40)
|
||||
copy(val, it.Value())
|
||||
entry = &syncDbEntry{key, val}
|
||||
// log.Trace(fmt.Sprintf("syncDb[%v/%v] - %v, batches: %v, total: %v, session total from db: %v/%v", sdb.key.Log(), sdb.priority, sdb.key.Log(), batches, total, sdb.dbTotal, sdb.total))
|
||||
more = fun(entry, sdb.quit)
|
||||
// log.Trace(fmt.Sprintf("syncDb[%v/%v] - %v, batches: %v, total: %v, session total from db: %v/%v", self.key.Log(), self.priority, self.key.Log(), batches, total, self.dbTotal, self.total))
|
||||
more = fun(entry, self.quit)
|
||||
if !more {
|
||||
// quit received when waiting to deliver entry, the entry will not be deleted
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] batch %v quit after %v/%v items", sdb.key.Log(), sdb.priority, batches, n, cnt))
|
||||
log.Trace(fmt.Sprintf("syncDb[%v/%v] batch %v quit after %v/%v items", self.key.Log(), self.priority, batches, n, cnt))
|
||||
break
|
||||
}
|
||||
// since subsequent batches of the same db session are indexed incrementally
|
||||
|
|
@ -326,22 +326,22 @@ func (sdb *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{},
|
|||
n++
|
||||
total++
|
||||
}
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] - db session closed, batches: %v, total: %v, session total from db: %v/%v", sdb.key.Log(), sdb.priority, batches, total, sdb.dbTotal, sdb.total))
|
||||
sdb.db.Write(del) // this could be async called only when db is idle
|
||||
log.Debug(fmt.Sprintf("syncDb[%v/%v] - db session closed, batches: %v, total: %v, session total from db: %v/%v", self.key.Log(), self.priority, batches, total, self.dbTotal, self.total))
|
||||
self.db.Write(del) // this could be async called only when db is idle
|
||||
it.Release()
|
||||
}
|
||||
}
|
||||
|
||||
//
|
||||
func (sdb *syncDb) stop() {
|
||||
close(sdb.quit)
|
||||
<-sdb.done
|
||||
func (self *syncDb) stop() {
|
||||
close(self.quit)
|
||||
<-self.done
|
||||
}
|
||||
|
||||
// calculate a dbkey for the request, for the db to work
|
||||
// see syncdb for db key structure
|
||||
// polimorphic: accepted types, see syncer#addRequest
|
||||
func (sdb *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDbEntry, err error) {
|
||||
func (self *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDbEntry, err error) {
|
||||
var key storage.Key
|
||||
var chunk *storage.Chunk
|
||||
var id uint64
|
||||
|
|
@ -367,7 +367,7 @@ func (sdb *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncD
|
|||
dbval := make([]byte, 40)
|
||||
|
||||
// encode key
|
||||
copy(dbkey[:], sdb.start[:34]) // db peer
|
||||
copy(dbkey[:], self.start[:34]) // db peer
|
||||
binary.BigEndian.PutUint64(dbkey[34:], counter)
|
||||
// encode value
|
||||
copy(dbval, key[:])
|
||||
|
|
|
|||
|
|
@ -71,25 +71,25 @@ func newTestSyncDb(priority, bufferSize, batchSize int, dbdir string, t *testing
|
|||
|
||||
}
|
||||
|
||||
func (db *testSyncDb) close() {
|
||||
db.db.Close()
|
||||
os.RemoveAll(db.dbdir)
|
||||
func (self *testSyncDb) close() {
|
||||
self.db.Close()
|
||||
os.RemoveAll(self.dbdir)
|
||||
}
|
||||
|
||||
func (db *testSyncDb) push(n int) {
|
||||
func (self *testSyncDb) push(n int) {
|
||||
for i := 0; i < n; i++ {
|
||||
db.buffer <- storage.Key(crypto.Keccak256([]byte{byte(db.c)}))
|
||||
db.sent = append(db.sent, db.c)
|
||||
db.c++
|
||||
self.buffer <- storage.Key(crypto.Keccak256([]byte{byte(self.c)}))
|
||||
self.sent = append(self.sent, self.c)
|
||||
self.c++
|
||||
}
|
||||
log.Debug(fmt.Sprintf("pushed %v requests", n))
|
||||
}
|
||||
|
||||
func (db *testSyncDb) draindb() {
|
||||
it := db.db.NewIterator()
|
||||
func (self *testSyncDb) draindb() {
|
||||
it := self.db.NewIterator()
|
||||
defer it.Release()
|
||||
for {
|
||||
it.Seek(db.start)
|
||||
it.Seek(self.start)
|
||||
if !it.Valid() {
|
||||
return
|
||||
}
|
||||
|
|
@ -98,44 +98,44 @@ func (db *testSyncDb) draindb() {
|
|||
return
|
||||
}
|
||||
it.Release()
|
||||
it = db.db.NewIterator()
|
||||
it = self.db.NewIterator()
|
||||
}
|
||||
}
|
||||
|
||||
func (db *testSyncDb) deliver(req interface{}, quit chan bool) bool {
|
||||
func (self *testSyncDb) deliver(req interface{}, quit chan bool) bool {
|
||||
_, db := req.(*syncDbEntry)
|
||||
key, _, _, _, err := parseRequest(req)
|
||||
if err != nil {
|
||||
db.t.Fatalf("unexpected error of key %v: %v", key, err)
|
||||
self.t.Fatalf("unexpected error of key %v: %v", key, err)
|
||||
}
|
||||
db.delivered = append(db.delivered, key)
|
||||
self.delivered = append(self.delivered, key)
|
||||
select {
|
||||
case db.fromDb <- db:
|
||||
case self.fromDb <- db:
|
||||
return true
|
||||
case <-quit:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (db *testSyncDb) expect(n int, db bool) {
|
||||
func (self *testSyncDb) expect(n int, db bool) {
|
||||
var ok bool
|
||||
// for n items
|
||||
for i := 0; i < n; i++ {
|
||||
ok = <-db.fromDb
|
||||
if db.at+1 > len(db.delivered) {
|
||||
db.t.Fatalf("expected %v, got %v", db.at+1, len(db.delivered))
|
||||
ok = <-self.fromDb
|
||||
if self.at+1 > len(self.delivered) {
|
||||
self.t.Fatalf("expected %v, got %v", self.at+1, len(self.delivered))
|
||||
}
|
||||
if len(db.sent) > db.at && !bytes.Equal(crypto.Keccak256([]byte{byte(db.sent[db.at])}), db.delivered[db.at]) {
|
||||
db.t.Fatalf("expected delivery %v/%v/%v to be hash of %v, from db: %v = %v", i, n, db.at, db.sent[db.at], ok, db)
|
||||
log.Debug(fmt.Sprintf("%v/%v/%v to be hash of %v, from db: %v = %v", i, n, db.at, db.sent[db.at], ok, db))
|
||||
if len(self.sent) > self.at && !bytes.Equal(crypto.Keccak256([]byte{byte(self.sent[self.at])}), self.delivered[self.at]) {
|
||||
self.t.Fatalf("expected delivery %v/%v/%v to be hash of %v, from db: %v = %v", i, n, self.at, self.sent[self.at], ok, db)
|
||||
log.Debug(fmt.Sprintf("%v/%v/%v to be hash of %v, from db: %v = %v", i, n, self.at, self.sent[self.at], ok, db))
|
||||
}
|
||||
if !ok && db {
|
||||
db.t.Fatalf("expected delivery %v/%v/%v from db", i, n, db.at)
|
||||
self.t.Fatalf("expected delivery %v/%v/%v from db", i, n, self.at)
|
||||
}
|
||||
if ok && !db {
|
||||
db.t.Fatalf("expected delivery %v/%v/%v from cache", i, n, db.at)
|
||||
self.t.Fatalf("expected delivery %v/%v/%v from cache", i, n, self.at)
|
||||
}
|
||||
db.at++
|
||||
self.at++
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue