diff --git a/swarm/api/filesystem_test.go b/swarm/api/filesystem_test.go index 8a15e735dc..2ee38336ce 100644 --- a/swarm/api/filesystem_test.go +++ b/swarm/api/filesystem_test.go @@ -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)) }) } diff --git a/swarm/api/storage_test.go b/swarm/api/storage_test.go index d260dd61d8..83a18bc5cc 100644 --- a/swarm/api/storage_test.go +++ b/swarm/api/storage_test.go @@ -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)) }) } diff --git a/swarm/network/syncdb.go b/swarm/network/syncdb.go index 491f417a44..c66d61b8c6 100644 --- a/swarm/network/syncdb.go +++ b/swarm/network/syncdb.go @@ -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[:]) diff --git a/swarm/network/syncdb_test.go b/swarm/network/syncdb_test.go index 7b70ee0f5e..be21d156f9 100644 --- a/swarm/network/syncdb_test.go +++ b/swarm/network/syncdb_test.go @@ -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++ } }