From 527c74902f4782fdffd029ebf2b6754349a02158 Mon Sep 17 00:00:00 2001 From: Eli Date: Wed, 2 May 2018 15:44:20 -0700 Subject: [PATCH] swarm/*: fixed naming capitalization errors --- swarm/fuse/fuse_file.go | 4 +- swarm/fuse/swarmfs_util.go | 6 +- swarm/network/kademlia/kademlia.go | 182 ++++++++++++------------ swarm/network/protocol.go | 2 +- swarm/network/syncdb.go | 124 ++++++++--------- swarm/network/syncer.go | 214 ++++++++++++++--------------- 6 files changed, 259 insertions(+), 273 deletions(-) diff --git a/swarm/fuse/fuse_file.go b/swarm/fuse/fuse_file.go index b0aa44ade4..2360f528aa 100644 --- a/swarm/fuse/fuse_file.go +++ b/swarm/fuse/fuse_file.go @@ -82,7 +82,7 @@ func (file *SwarmFile) Attr(ctx context.Context, a *fuse.Attr) error { a.Gid = uint32(os.Getegid()) if file.fileSize == -1 { - reader := file.mountInfo.swarmApi.Retrieve(file.key) + reader := file.mountInfo.swarmAPI.Retrieve(file.key) quitC := make(chan bool) size, err := reader.Size(quitC) if err != nil { @@ -99,7 +99,7 @@ func (file *SwarmFile) Read(ctx context.Context, req *fuse.ReadRequest, resp *fu file.lock.RLock() defer file.lock.RUnlock() if file.reader == nil { - file.reader = file.mountInfo.swarmApi.Retrieve(file.key) + file.reader = file.mountInfo.swarmAPI.Retrieve(file.key) } buf := make([]byte, req.Size) n, err := file.reader.ReadAt(buf, req.Offset) diff --git a/swarm/fuse/swarmfs_util.go b/swarm/fuse/swarmfs_util.go index 169b674876..6b8e3f3f82 100644 --- a/swarm/fuse/swarmfs_util.go +++ b/swarm/fuse/swarmfs_util.go @@ -47,7 +47,7 @@ func externalUnmount(mountPoint string) error { } func addFileToSwarm(sf *SwarmFile, content []byte, size int) error { - fkey, mhash, err := sf.mountInfo.swarmApi.AddFile(sf.mountInfo.LatestManifest, sf.path, sf.name, content, true) + fkey, mhash, err := sf.mountInfo.swarmAPI.AddFile(sf.mountInfo.LatestManifest, sf.path, sf.name, content, true) if err != nil { return err } @@ -66,7 +66,7 @@ func addFileToSwarm(sf *SwarmFile, content []byte, size int) error { } func removeFileFromSwarm(sf *SwarmFile) error { - mkey, err := sf.mountInfo.swarmApi.RemoveFile(sf.mountInfo.LatestManifest, sf.path, sf.name, true) + mkey, err := sf.mountInfo.swarmAPI.RemoveFile(sf.mountInfo.LatestManifest, sf.path, sf.name, true) if err != nil { return err } @@ -102,7 +102,7 @@ func removeDirectoryFromSwarm(sd *SwarmDir) error { } func appendToExistingFileInSwarm(sf *SwarmFile, content []byte, offset int64, length int64) error { - fkey, mhash, err := sf.mountInfo.swarmApi.AppendFile(sf.mountInfo.LatestManifest, sf.path, sf.name, sf.fileSize, content, sf.key, offset, length, true) + fkey, mhash, err := sf.mountInfo.swarmAPI.AppendFile(sf.mountInfo.LatestManifest, sf.path, sf.name, sf.fileSize, content, sf.key, offset, length, true) if err != nil { return err } diff --git a/swarm/network/kademlia/kademlia.go b/swarm/network/kademlia/kademlia.go index 92377ee2a1..d849f3ca9f 100644 --- a/swarm/network/kademlia/kademlia.go +++ b/swarm/network/kademlia/kademlia.go @@ -109,31 +109,31 @@ func New(addr Address, params *KadParams) *Kademlia { } // accessor for KAD base address -func (k *Kademlia) Addr() Address { - return k.addr +func (kad *Kademlia) Addr() Address { + return kad.addr } // accessor for KAD active node count -func (k *Kademlia) Count() int { - defer k.lock.Unlock() - k.lock.Lock() - return k.count +func (kad *Kademlia) Count() int { + defer kad.lock.Unlock() + kad.lock.Lock() + return kad.count } // accessor for KAD active node count -func (k *Kademlia) DBCount() int { - return k.db.count() +func (kad *Kademlia) DBCount() int { + return kad.db.count() } // On is the entry point called when a new nodes is added // unsafe in that node is not checked to be already active node (to be called once) -func (k *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error) { - log.Debug(fmt.Sprintf("%v", k)) - defer k.lock.Unlock() - k.lock.Lock() +func (kad *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error) { + log.Debug(fmt.Sprintf("%v", kad)) + defer kad.lock.Unlock() + kad.lock.Lock() - index := k.proximityBin(node.Addr()) - record := k.db.findOrCreate(index, node.Addr(), node.Url()) + index := kad.proximityBin(node.Addr()) + record := kad.db.findOrCreate(index, node.Addr(), node.Url()) if cb != nil { err = cb(record, node) @@ -145,21 +145,21 @@ func (k *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error) { } // insert in kademlia table of active nodes - bucket := k.buckets[index] + bucket := kad.buckets[index] // if bucket is full insertion replaces the worst node // TODO: give priority to peers with active traffic - if len(bucket) < k.BucketSize { // >= allows us to add peers beyond the bucketsize limitation - k.buckets[index] = append(bucket, node) + if len(bucket) < kad.BucketSize { // >= allows us to add peers beyond the bucketsize limitation + kad.buckets[index] = append(bucket, node) bucketAddIndexCount[index].Inc(1) log.Debug(fmt.Sprintf("add node %v to table", node)) - k.setProxLimit(index, true) + kad.setProxLimit(index, true) record.node = node - k.count++ + kad.count++ return nil } // always rotate peers - idle := k.MaxIdleInterval + idle := kad.MaxIdleInterval var pos int var replaced Node for i, p := range bucket { @@ -174,41 +174,41 @@ func (k *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error) { log.Debug(fmt.Sprintf("all peers wanted, PO%03d bucket full", index)) return fmt.Errorf("bucket full") } - log.Debug(fmt.Sprintf("node %v replaced by %v (idle for %v > %v)", replaced, node, idle, k.MaxIdleInterval)) + log.Debug(fmt.Sprintf("node %v replaced by %v (idle for %v > %v)", replaced, node, idle, kad.MaxIdleInterval)) replaced.Drop() // actually replace in the row. When off(node) is called, the peer is no longer in the row bucket[pos] = node // there is no change in bucket cardinalities so no prox limit adjustment is needed record.node = node - k.count++ + kad.count++ return nil } // Off is the called when a node is taken offline (from the protocol main loop exit) -func (k *Kademlia) Off(node Node, cb func(*NodeRecord, Node)) (err error) { - k.lock.Lock() - defer k.lock.Unlock() +func (kad *Kademlia) Off(node Node, cb func(*NodeRecord, Node)) (err error) { + kad.lock.Lock() + defer kad.lock.Unlock() - index := k.proximityBin(node.Addr()) + index := kad.proximityBin(node.Addr()) bucketRmIndexCount[index].Inc(1) - bucket := k.buckets[index] + bucket := kad.buckets[index] for i := 0; i < len(bucket); i++ { if node.Addr() == bucket[i].Addr() { - k.buckets[index] = append(bucket[:i], bucket[(i+1):]...) - k.setProxLimit(index, false) + kad.buckets[index] = append(bucket[:i], bucket[(i+1):]...) + kad.setProxLimit(index, false) break } } - record := k.db.index[node.Addr()] + record := kad.db.index[node.Addr()] // callback on remove if cb != nil { cb(record, record.node) } record.node = nil - k.count-- - log.Debug(fmt.Sprintf("remove node %v from table, population now is %v", node, k.count)) + kad.count-- + log.Debug(fmt.Sprintf("remove node %v from table, population now is %v", node, kad.count)) return } @@ -218,39 +218,39 @@ func (k *Kademlia) Off(node Node, cb func(*NodeRecord, Node)) (err error) { // 2) the sum of all items are the minimum possible but higher than ProxBinSize // adjust Prox (proxLimit and proxSize after an insertion/removal of nodes) // caller holds the lock -func (k *Kademlia) setProxLimit(r int, on bool) { +func (kad *Kademlia) setProxLimit(r int, on bool) { // if the change is outside the core (PO lower) // and the change does not leave a bucket empty then // no adjustment needed - if r < k.proxLimit && len(k.buckets[r]) > 0 { + if r < kad.proxLimit && len(kad.buckets[r]) > 0 { return } // if on=a node was added, then r must be within prox limit so increment cardinality if on { - k.proxSize++ - curr := len(k.buckets[k.proxLimit]) + kad.proxSize++ + curr := len(kad.buckets[kad.proxLimit]) // if now core is big enough without the furthest bucket, then contract // this can result in more than one bucket change - for k.proxSize >= k.ProxBinSize+curr && curr > 0 { - k.proxSize -= curr - k.proxLimit++ - curr = len(k.buckets[k.proxLimit]) + for kad.proxSize >= kad.ProxBinSize+curr && curr > 0 { + kad.proxSize -= curr + kad.proxLimit++ + curr = len(kad.buckets[kad.proxLimit]) - log.Trace(fmt.Sprintf("proxbin contraction (size: %v, limit: %v, bin: %v)", k.proxSize, k.proxLimit, r)) + log.Trace(fmt.Sprintf("proxbin contraction (size: %v, limit: %v, bin: %v)", kad.proxSize, kad.proxLimit, r)) } return } // otherwise - if r >= k.proxLimit { - k.proxSize-- + if r >= kad.proxLimit { + kad.proxSize-- } // expand core by lowering prox limit until hit zero or cover the empty bucket or reached target cardinality - for (k.proxSize < k.ProxBinSize || r < k.proxLimit) && - k.proxLimit > 0 { + for (kad.proxSize < kad.ProxBinSize || r < kad.proxLimit) && + kad.proxLimit > 0 { // - k.proxLimit-- - k.proxSize += len(k.buckets[k.proxLimit]) - log.Trace(fmt.Sprintf("proxbin expansion (size: %v, limit: %v, bin: %v)", k.proxSize, k.proxLimit, r)) + kad.proxLimit-- + kad.proxSize += len(kad.buckets[kad.proxLimit]) + log.Trace(fmt.Sprintf("proxbin expansion (size: %v, limit: %v, bin: %v)", kad.proxSize, kad.proxLimit, r)) } } @@ -259,15 +259,15 @@ FindClosest returns the list of nodes belonging to the same proximity bin as the target. The most proximate bin will be the union of the bins between proxLimit and MaxProx. */ -func (k *Kademlia) FindClosest(target Address, max int) []Node { - k.lock.Lock() - defer k.lock.Unlock() +func (kad *Kademlia) FindClosest(target Address, max int) []Node { + kad.lock.Lock() + defer kad.lock.Unlock() r := nodesByDistance{ target: target, } - po := k.proximityBin(target) + po := kad.proximityBin(target) index := po step := 1 log.Trace(fmt.Sprintf("serving %v nodes at %v (PO%02d)", max, index, po)) @@ -284,17 +284,17 @@ func (k *Kademlia) FindClosest(target Address, max int) []Node { var n int for index >= 0 { // add entire bucket - for _, p := range k.buckets[index] { + for _, p := range kad.buckets[index] { r.push(p, limit) n++ } // terminate if index reached the bottom or enough peers > min - log.Trace(fmt.Sprintf("add %v -> %v (PO%02d, PO%03d)", len(k.buckets[index]), n, index, po)) + log.Trace(fmt.Sprintf("add %v -> %v (PO%02d, PO%03d)", len(kad.buckets[index]), n, index, po)) if n >= min && (step < 0 || max == 0) { break } // reach top most non-empty PO bucket, turn around - if index == k.MaxProx { + if index == kad.MaxProx { index = po step = -1 } @@ -304,15 +304,15 @@ func (k *Kademlia) FindClosest(target Address, max int) []Node { return r.nodes } -func (k *Kademlia) Suggest() (*NodeRecord, bool, int) { - defer k.lock.RUnlock() - k.lock.RLock() - return k.db.findBest(k.BucketSize, func(i int) int { return len(k.buckets[i]) }) +func (kad *Kademlia) Suggest() (*NodeRecord, bool, int) { + defer kad.lock.RUnlock() + kad.lock.RLock() + return kad.db.findBest(kad.BucketSize, func(i int) int { return len(kad.buckets[i]) }) } // Add node records to kaddb (persisted node record db) -func (k *Kademlia) Add(nrs []*NodeRecord) { - k.db.add(nrs, k.proximityBin) +func (kad *Kademlia) Add(nrs []*NodeRecord) { + kad.db.add(nrs, kad.proximityBin) } // nodesByDistance is a list of nodes, ordered by distance to target. @@ -369,52 +369,52 @@ a guaranteed constant maximum limit on the number of hops needed to reach one node from the other. */ -func (k *Kademlia) proximityBin(other Address) (ret int) { - ret = proximity(k.addr, other) - if ret > k.MaxProx { - ret = k.MaxProx +func (kad *Kademlia) proximityBin(other Address) (ret int) { + ret = proximity(kad.addr, other) + if ret > kad.MaxProx { + ret = kad.MaxProx } return } // provides keyrange for chunk db iteration -func (k *Kademlia) KeyRange(other Address) (start, stop Address) { - defer k.lock.RUnlock() - k.lock.RLock() - return KeyRange(k.addr, other, k.proxLimit) +func (kad *Kademlia) KeyRange(other Address) (start, stop Address) { + defer kad.lock.RUnlock() + kad.lock.RLock() + return KeyRange(kad.addr, other, kad.proxLimit) } // save persists kaddb on disk (written to file on path in json format. -func (k *Kademlia) Save(path string, cb func(*NodeRecord, Node)) error { - return k.db.save(path, cb) +func (kad *Kademlia) Save(path string, cb func(*NodeRecord, Node)) error { + return kad.db.save(path, cb) } // Load(path) loads the node record database (kaddb) from file on path. -func (k *Kademlia) Load(path string, cb func(*NodeRecord, Node) error) (err error) { - return k.db.load(path, cb) +func (kad *Kademlia) Load(path string, cb func(*NodeRecord, Node) error) (err error) { + return kad.db.load(path, cb) } // kademlia table + kaddb table displayed with ascii -func (k *Kademlia) String() string { - defer k.lock.RUnlock() - k.lock.RLock() - defer k.db.lock.RUnlock() - k.db.lock.RLock() +func (kad *Kademlia) String() string { + defer kad.lock.RUnlock() + kad.lock.RLock() + defer kad.db.lock.RUnlock() + kad.db.lock.RLock() var rows []string rows = append(rows, "=========================================================================") - rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), k.addr.String()[:6])) - rows = append(rows, fmt.Sprintf("population: %d (%d), proxLimit: %d, proxSize: %d", k.count, len(k.db.index), k.proxLimit, k.proxSize)) - rows = append(rows, fmt.Sprintf("MaxProx: %d, ProxBinSize: %d, BucketSize: %d", k.MaxProx, k.ProxBinSize, k.BucketSize)) + rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), kad.addr.String()[:6])) + rows = append(rows, fmt.Sprintf("population: %d (%d), proxLimit: %d, proxSize: %d", kad.count, len(kad.db.index), kad.proxLimit, kad.proxSize)) + rows = append(rows, fmt.Sprintf("MaxProx: %d, ProxBinSize: %d, BucketSize: %d", kad.MaxProx, kad.ProxBinSize, kad.BucketSize)) - for i, bucket := range k.buckets { + for i, bucket := range kad.buckets { - if i == k.proxLimit { + if i == kad.proxLimit { rows = append(rows, fmt.Sprintf("============ PROX LIMIT: %d ==========================================", i)) } row := []string{fmt.Sprintf("%03d", i), fmt.Sprintf("%2d", len(bucket))} var k int - c := k.db.cursors[i] + c := kad.db.cursors[i] for ; k < len(bucket); k++ { p := bucket[(c+k)%len(bucket)] row = append(row, p.Addr().String()[:6]) @@ -425,16 +425,16 @@ func (k *Kademlia) String() string { for ; k < 4; k++ { row = append(row, " ") } - row = append(row, fmt.Sprintf("| %2d %2d", len(k.db.Nodes[i]), k.db.cursors[i])) + row = append(row, fmt.Sprintf("| %2d %2d", len(kad.db.Nodes[i]), kad.db.cursors[i])) - for j, p := range k.db.Nodes[i] { + for j, p := range kad.db.Nodes[i] { row = append(row, p.Addr.String()[:6]) if j == 3 { break } } rows = append(rows, strings.Join(row, " ")) - if i == k.MaxProx { + if i == kad.MaxProx { } } rows = append(rows, "=========================================================================") @@ -442,12 +442,12 @@ func (k *Kademlia) String() string { } //We have to build up the array of counters for each index -func (k *Kademlia) initMetricsVariables() { +func (kad *Kademlia) initMetricsVariables() { //create the arrays - bucketAddIndexCount = make([]metrics.Counter, k.MaxProx+1) - bucketRmIndexCount = make([]metrics.Counter, k.MaxProx+1) + bucketAddIndexCount = make([]metrics.Counter, kad.MaxProx+1) + bucketRmIndexCount = make([]metrics.Counter, kad.MaxProx+1) //at each index create a metrics counter - for i := 0; i < (k.KadParams.MaxProx + 1); i++ { + for i := 0; i < (kad.KadParams.MaxProx + 1); i++ { bucketAddIndexCount[i] = metrics.NewRegisteredCounter(fmt.Sprintf("network.kademlia.bucket.add.%d.index", i), nil) bucketRmIndexCount[i] = metrics.NewRegisteredCounter(fmt.Sprintf("network.kademlia.bucket.rm.%d.index", i), nil) } diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 4dadbf6d61..92539ad526 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -229,7 +229,7 @@ func (bzz *bzz) handle() error { return fmt.Errorf("<- %v: Data too short (%v)", msg, n) } // last Active time is set only when receiving chunks - self.lastActive = time.Now() + bzz.lastActive = time.Now() log.Trace(fmt.Sprintf("incoming store request: %s", req.String())) // swap accounting is done within forwarding bzz.storage.HandleStoreRequestMsg(&req, &peer{bzz: bzz}) diff --git a/swarm/network/syncdb.go b/swarm/network/syncdb.go index 2ef2bbf986..491f417a44 100644 --- a/swarm/network/syncdb.go +++ b/swarm/network/syncdb.go @@ -31,13 +31,11 @@ const counterKeyPrefix = 0x01 /* syncDb is a queueing service for outgoing deliveries. One instance per priority queue for each peer - a syncDb instance maintains an in-memory buffer (of capacity bufferSize) once its in-memory buffer is full it switches to persisting in db and dbRead iterator iterates through the items keeping their order once the db read catches up (there is no more items in the db) then it switches back to in-memory buffer. - when syncdb is stopped all items in the buffer are saved to the db */ type syncDb struct { @@ -89,26 +87,21 @@ func newSyncDb(db *storage.LDBDatabase, key storage.Key, priority uint, bufferSi /* bufferRead is a forever iterator loop that takes care of delivering outgoing store requests reads from incoming buffer - its argument is the deliver function taking the item as first argument and a quit channel as second. Closing of this channel is supposed to abort all waiting for delivery (typically network write) - The iteration switches between 2 modes, * buffer mode reads the in-memory buffer and delivers the items directly * db mode reads from the buffer and writes to the db, parallelly another routine is started that reads from the db and delivers items - If there is buffer contention in buffer mode (slow network, high upload volume) syncdb switches to db mode and starts dbRead 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 (db *syncDb) bufferRead(deliver func(interface{}, chan bool) bool) { +func (sdb *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{} @@ -116,18 +109,18 @@ func (db *syncDb) bufferRead(deliver func(interface{}, chan bool) bool) { var inBatch, inDb int batch := new(leveldb.Batch) var dbSize chan int - quit := db.quit + quit := sdb.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 := db.db.Get(db.counterKey) + data, err := sdb.db.Get(sdb.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", db.key.Log(), db.priority, counter)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter read from db at %v", sdb.key.Log(), sdb.priority, counter)) } else { - log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter starts at %v", db.key.Log(), db.priority, counter)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] - counter starts at %v", sdb.key.Log(), sdb.priority, counter)) } LOOP: @@ -139,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, db.quit) + more = deliver(req, sdb.quit) if !more { - log.Debug(fmt.Sprintf("syncDb[%v/%v] quit: switching to db. session tally (db/total): %v/%v", db.key.Log(), db.priority, db.dbTotal, db.total)) + 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)) // received quit signal, save request currently waiting delivery // by switching to db mode and closing the buffer buffer = nil - db = db.buffer + db = sdb.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 } - db.total++ - log.Trace(fmt.Sprintf("syncDb[%v/%v] deliver (db/total): %v/%v", db.key.Log(), db.priority, db.dbTotal, db.total)) + sdb.total++ + log.Trace(fmt.Sprintf("syncDb[%v/%v] deliver (db/total): %v/%v", sdb.key.Log(), sdb.priority, sdb.dbTotal, sdb.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", db.key.Log(), db.priority, cap(buffer), db.dbTotal, db.total)) + 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)) buffer = nil - db = db.buffer + db = sdb.buffer } continue LOOP @@ -167,30 +160,30 @@ LOOP: if !more { // only if quit is called, saved all the buffer binary.BigEndian.PutUint64(counterValue, counter) - batch.Put(db.counterKey, counterValue) // persist counter in batch - db.writeSyncBatch(batch) // save batch - log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save current batch to db", db.key.Log(), db.priority)) + 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)) break LOOP } - db.dbTotal++ - db.total++ + sdb.dbTotal++ + sdb.total++ // otherwise break after select - case dbSize = <-db.batch: + case dbSize = <-sdb.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", db.key.Log(), db.priority)) + log.Debug(fmt.Sprintf("syncDb[%v/%v] empty db: switching to buffer", sdb.key.Log(), sdb.priority)) db = nil - buffer = db.buffer + buffer = sdb.buffer dbSize <- 0 // indicates to 'caller' that batch has been written inDb = 0 continue LOOP } binary.BigEndian.PutUint64(counterValue, counter) - batch.Put(db.counterKey, counterValue) - log.Debug(fmt.Sprintf("syncDb[%v/%v] write batch %v/%v - %x - %x", db.key.Log(), db.priority, inBatch, counter, db.counterKey, counterValue)) - batch = db.writeSyncBatch(batch) + 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) dbSize <- inBatch // indicates to 'caller' that batch has been written inBatch = 0 continue LOOP @@ -198,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 = db.buffer + db = sdb.buffer buffer = nil quit = nil - log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save buffer to db", db.key.Log(), db.priority)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save buffer to db", sdb.key.Log(), sdb.priority)) close(db) continue LOOP } // only get here if we put req into db - entry, err = db.newSyncDbEntry(req, counter) + entry, err = sdb.newSyncDbEntry(req, counter) if err != nil { - log.Warn(fmt.Sprintf("syncDb[%v/%v] saving request %v (#%v/%v) failed: %v", db.key.Log(), db.priority, req, inBatch, inDb, err)) + log.Warn(fmt.Sprintf("syncDb[%v/%v] saving request %v (#%v/%v) failed: %v", sdb.key.Log(), sdb.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)", db.key.Log(), db.priority, req, entry, inBatch, inDb, counter)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] to batch %v '%v' (#%v/%v/%v)", sdb.key.Log(), sdb.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", db.key.Log(), db.priority)) - go db.dbRead(true, counter, deliver) + log.Trace(fmt.Sprintf("syncDb[%v/%v] start dbRead", sdb.key.Log(), sdb.priority)) + go sdb.dbRead(true, counter, deliver) } inDb++ inBatch++ counter++ // need to save the batch if it gets too large (== dbBatchSize) - if inBatch%int(db.dbBatchSize) == 0 { - batch = db.writeSyncBatch(batch) + if inBatch%int(sdb.dbBatchSize) == 0 { + batch = sdb.writeSyncBatch(batch) } } - log.Info(fmt.Sprintf("syncDb[%v:%v]: saved %v keys (saved counter at %v)", db.key.Log(), db.priority, inBatch, counter)) - close(db.done) + log.Info(fmt.Sprintf("syncDb[%v:%v]: saved %v keys (saved counter at %v)", sdb.key.Log(), sdb.priority, inBatch, counter)) + close(sdb.done) } // writes the batch to the db and returns a new batch object -func (db *syncDb) writeSyncBatch(batch *leveldb.Batch) *leveldb.Batch { - err := db.db.Write(batch) +func (sdb *syncDb) writeSyncBatch(batch *leveldb.Batch) *leveldb.Batch { + err := sdb.db.Write(batch) if err != nil { - log.Warn(fmt.Sprintf("syncDb[%v/%v] saving batch to db failed: %v", db.key.Log(), db.priority, err)) + log.Warn(fmt.Sprintf("syncDb[%v/%v] saving batch to db failed: %v", sdb.key.Log(), sdb.priority, err)) return batch } return new(leveldb.Batch) @@ -256,25 +249,22 @@ func (entry syncDbEntry) String() string { this is mainly to prevent crashes due to network output buffer contention (???) as well as to make syncronisation resilient to disconnects the messages are supposed to be sent in the p2p priority queue. - the request DB is shared between peers, but domains for each syncdb are disjoint. dbkeys (42 bytes) are structured: * 0: 0x00 (0x01 reserved for counter key) * 1: priorities - priority (so that high priority can be replayed first) * 2-33: peers address * 34-41: syncdb counter to preserve order (this field is missing for the counter key) - values (40 bytes) are: * 0-31: key * 32-39: request id - 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 (db *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{}, chan bool) bool) { +func (sdb *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{}, chan bool) bool) { key := make([]byte, 42) - copy(key, db.start) + copy(key, sdb.start) binary.BigEndian.PutUint64(key[34:], counter) var batches, n, cnt, total int var more bool @@ -290,8 +280,8 @@ func (db *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 db.batch <- batchSizes: - case <-db.quit: + case sdb.batch <- batchSizes: + case <-sdb.quit: return } // wait for the write to finish and get the item count in the next batch @@ -302,31 +292,31 @@ func (db *syncDb) dbRead(useBatches bool, counter uint64, fun func(interface{}, return } } - it = db.db.NewIterator() + it = sdb.db.NewIterator() it.Seek(key) if !it.Valid() { - copy(key, db.start) + copy(key, sdb.start) useBatches = true continue } del = new(leveldb.Batch) - log.Trace(fmt.Sprintf("syncDb[%v/%v]: new iterator: %x (batch %v, count %v)", db.key.Log(), db.priority, key, batches, cnt)) + log.Trace(fmt.Sprintf("syncDb[%v/%v]: new iterator: %x (batch %v, count %v)", sdb.key.Log(), sdb.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, db.start) + copy(key, sdb.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", db.key.Log(), db.priority, db.key.Log(), batches, total, db.dbTotal, db.total)) - more = fun(entry, db.quit) + // 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) 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", db.key.Log(), db.priority, batches, n, cnt)) + log.Trace(fmt.Sprintf("syncDb[%v/%v] batch %v quit after %v/%v items", sdb.key.Log(), sdb.priority, batches, n, cnt)) break } // since subsequent batches of the same db session are indexed incrementally @@ -336,22 +326,22 @@ func (db *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", db.key.Log(), db.priority, batches, total, db.dbTotal, db.total)) - db.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", 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 it.Release() } } // -func (db *syncDb) stop() { - close(db.quit) - <-db.done +func (sdb *syncDb) stop() { + close(sdb.quit) + <-sdb.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 (db *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDbEntry, err error) { +func (sdb *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDbEntry, err error) { var key storage.Key var chunk *storage.Chunk var id uint64 @@ -359,10 +349,10 @@ func (db *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDb var sreq *storeRequestMsgData if key, ok = req.(storage.Key); ok { - id = generateId() + id = generateID() } else if chunk, ok = req.(*storage.Chunk); ok { key = chunk.Key - id = generateId() + id = generateID() } else if sreq, ok = req.(*storeRequestMsgData); ok { key = sreq.Key id = sreq.Id @@ -377,7 +367,7 @@ func (db *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDb dbval := make([]byte, 40) // encode key - copy(dbkey[:], db.start[:34]) // db peer + copy(dbkey[:], sdb.start[:34]) // db peer binary.BigEndian.PutUint64(dbkey[34:], counter) // encode value copy(dbval, key[:]) diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index 8b5c6f80df..7ba7374075 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -262,24 +262,23 @@ func decodeSync(meta *json.RawMessage) (*syncState, error) { * if Last < LastSeenAt then all items in between then process all backlog from upto last disconnect * if Last > 0 && - sync is called from the syncer constructor and is not supposed to be used externally */ -func (sync *syncer) sync() { - state := sync.state +func (s *syncer) sync() { + state := s.state // sync finished - defer close(sync.syncStates) + defer close(s.syncStates) // 0. first replay stale requests from request db if state.SessionAt == 0 { - log.Debug(fmt.Sprintf("syncer[%v]: nothing to sync", sync.key.Log())) + log.Debug(fmt.Sprintf("syncer[%v]: nothing to sync", s.key.Log())) return } - log.Debug(fmt.Sprintf("syncer[%v]: start replaying stale requests from request db", sync.key.Log())) + log.Debug(fmt.Sprintf("syncer[%v]: start replaying stale requests from request db", s.key.Log())) for p := priorities - 1; p >= 0; p-- { - sync.queues[p].dbRead(false, 0, sync.replay()) + s.queues[p].dbRead(false, 0, s.replay()) } - log.Debug(fmt.Sprintf("syncer[%v]: done replaying stale requests from request db", sync.key.Log())) + log.Debug(fmt.Sprintf("syncer[%v]: done replaying stale requests from request db", s.key.Log())) // unless peer is synced sync unfinished history beginning on if !state.Synced { @@ -288,9 +287,9 @@ func (sync *syncer) sync() { if !storage.IsZeroKey(state.Latest) { // 1. there is unfinished earlier sync state.Start = state.Latest - log.Debug(fmt.Sprintf("syncer[%v]: start syncronising backlog (unfinished sync: %v)", sync.key.Log(), state)) + log.Debug(fmt.Sprintf("syncer[%v]: start syncronising backlog (unfinished sync: %v)", s.key.Log(), state)) // blocks while the entire history upto state is synced - sync.syncState(state) + s.syncState(state) if state.Last < state.SessionAt { state.First = state.Last + 1 } @@ -300,8 +299,8 @@ func (sync *syncer) sync() { // 2. sync up to last disconnect1 if state.First < state.LastSeenAt { state.Last = state.LastSeenAt - log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history upto last disconnect at %v: %v", sync.key.Log(), state.LastSeenAt, state)) - sync.syncState(state) + log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history upto last disconnect at %v: %v", s.key.Log(), state.LastSeenAt, state)) + s.syncState(state) state.First = state.LastSeenAt } state.Latest = storage.ZeroKey @@ -315,28 +314,28 @@ func (sync *syncer) sync() { // if there have been new chunks since last session if state.LastSeenAt < state.SessionAt { state.Last = state.SessionAt - log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history since last disconnect at %v up until session start at %v: %v", sync.key.Log(), state.LastSeenAt, state.SessionAt, state)) + log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history since last disconnect at %v up until session start at %v: %v", s.key.Log(), state.LastSeenAt, state.SessionAt, state)) // blocks until state syncing is finished - sync.syncState(state) + s.syncState(state) } - log.Info(fmt.Sprintf("syncer[%v]: syncing all history complete", sync.key.Log())) + log.Info(fmt.Sprintf("syncer[%v]: syncing all history complete", s.key.Log())) } // wait till syncronised block uptil state is synced -func (sync *syncer) syncState(state *syncState) { - sync.syncStates <- state +func (s *syncer) syncState(state *syncState) { + s.syncStates <- state select { case <-state.synced: - case <-sync.quit: + case <-s.quit: } } // stop quits both request processor and saves the request cache to disk -func (sync *syncer) stop() { - close(sync.quit) - log.Trace(fmt.Sprintf("syncer[%v]: stop and save sync request db backlog", sync.key.Log())) - for _, db := range sync.queues { +func (s *syncer) stop() { + close(s.quit) + log.Trace(fmt.Sprintf("syncer[%v]: stop and save sync request db backlog", s.key.Log())) + for _, db := range s.queues { db.stop() } } @@ -351,7 +350,7 @@ func (req *syncRequest) String() string { return fmt.Sprintf("", req.Key.Log(), req.Priority) } -func (sync *syncer) newSyncRequest(req interface{}, p int) (*syncRequest, error) { +func (s *syncer) newSyncRequest(req interface{}, p int) (*syncRequest, error) { key, _, _, _, err := parseRequest(req) // TODO: if req has chunk, it should be put in a cache // create @@ -365,11 +364,11 @@ func (sync *syncer) newSyncRequest(req interface{}, p int) (*syncRequest, error) // * read is on demand, blocking unless history channel is read // * accepts sync requests (syncStates) to create new db iterator // * closes the channel one iteration finishes -func (sync *syncer) syncHistory(state *syncState) chan interface{} { +func (s *syncer) syncHistory(state *syncState) chan interface{} { var n uint history := make(chan interface{}) - log.Debug(fmt.Sprintf("syncer[%v]: syncing history between %v - %v for chunk addresses %v - %v", sync.key.Log(), state.First, state.Last, state.Start, state.Stop)) - it := sync.dbAccess.iterator(state) + log.Debug(fmt.Sprintf("syncer[%v]: syncing history between %v - %v for chunk addresses %v - %v", s.key.Log(), state.First, state.Last, state.Start, state.Stop)) + it := s.dbAccess.iterator(state) if it != nil { go func() { // signal end of the iteration ended @@ -384,22 +383,22 @@ func (sync *syncer) syncHistory(state *syncState) chan interface{} { // blocking until history channel is read from case history <- key: n++ - log.Trace(fmt.Sprintf("syncer[%v]: history: %v (%v keys)", sync.key.Log(), key.Log(), n)) + log.Trace(fmt.Sprintf("syncer[%v]: history: %v (%v keys)", s.key.Log(), key.Log(), n)) state.Latest = key - case <-sync.quit: + case <-s.quit: return } } - log.Debug(fmt.Sprintf("syncer[%v]: finished syncing history between %v - %v for chunk addresses %v - %v (at %v) (chunks = %v)", sync.key.Log(), state.First, state.Last, state.Start, state.Stop, state.Latest, n)) + log.Debug(fmt.Sprintf("syncer[%v]: finished syncing history between %v - %v for chunk addresses %v - %v (at %v) (chunks = %v)", s.key.Log(), state.First, state.Last, state.Start, state.Stop, state.Latest, n)) }() } return history } // triggers key syncronisation -func (sync *syncer) sendUnsyncedKeys() { +func (s *syncer) sendUnsyncedKeys() { select { - case sync.deliveryRequest <- true: + case s.deliveryRequest <- true: default: } } @@ -410,7 +409,7 @@ func (sync *syncer) sendUnsyncedKeys() { // historical data is used so historical items are lower priority within // their priority group. // * Order of historical data is unspecified -func (sync *syncer) syncUnsyncedKeys() { +func (s *syncer) syncUnsyncedKeys() { // send out new var unsynced []*syncRequest var more, justSynced bool @@ -418,12 +417,12 @@ func (sync *syncer) syncUnsyncedKeys() { var history chan interface{} priority := High - keys := sync.keys[priority] + keys := s.keys[priority] var newUnsyncedKeys, deliveryRequest chan bool keyCounts := make([]int, priorities) - histPrior := sync.SyncPriorities[HistoryReq] - syncStates := sync.syncStates - state := sync.state + histPrior := s.SyncPriorities[HistoryReq] + syncStates := s.syncStates + state := s.state LOOP: for { @@ -439,15 +438,15 @@ LOOP: PRIORITIES: for priority = High; priority >= 0; priority-- { // the first priority channel that is non-empty will be assigned to keys - if len(sync.keys[priority]) > 0 { - log.Trace(fmt.Sprintf("syncer[%v]: reading request with priority %v", sync.key.Log(), priority)) - keys = sync.keys[priority] + if len(s.keys[priority]) > 0 { + log.Trace(fmt.Sprintf("syncer[%v]: reading request with priority %v", s.key.Log(), priority)) + keys = s.keys[priority] break PRIORITIES } - log.Trace(fmt.Sprintf("syncer[%v/%v]: queue: [%v, %v, %v]", sync.key.Log(), priority, len(sync.keys[High]), len(sync.keys[Medium]), len(sync.keys[Low]))) + log.Trace(fmt.Sprintf("syncer[%v/%v]: queue: [%v, %v, %v]", s.key.Log(), priority, len(s.keys[High]), len(s.keys[Medium]), len(s.keys[Low]))) // if the input queue is empty on this level, resort to history if there is any if uint(priority) == histPrior && history != nil { - log.Trace(fmt.Sprintf("syncer[%v]: reading history for %v", sync.key.Log(), sync.key)) + log.Trace(fmt.Sprintf("syncer[%v]: reading history for %v", s.key.Log(), s.key)) keys = history break PRIORITIES } @@ -457,8 +456,8 @@ LOOP: // if peer ready to receive but nothing to send if keys == nil && deliveryRequest == nil { // if no items left and switch to waiting mode - log.Trace(fmt.Sprintf("syncer[%v]: buffers consumed. Waiting", sync.key.Log())) - newUnsyncedKeys = sync.newUnsyncedKeys + log.Trace(fmt.Sprintf("syncer[%v]: buffers consumed. Waiting", s.key.Log())) + newUnsyncedKeys = s.newUnsyncedKeys } // send msg iff @@ -469,48 +468,48 @@ LOOP: if deliveryRequest == nil && (justSynced || len(unsynced) > 0 && keys == nil || - len(unsynced) == int(sync.SyncBatchSize)) { + len(unsynced) == int(s.SyncBatchSize)) { justSynced = false // listen to requests - deliveryRequest = sync.deliveryRequest + deliveryRequest = s.deliveryRequest newUnsyncedKeys = nil // not care about data until next req comes in // set sync to current counter // (all nonhistorical outgoing traffic sheduled and persisted - state.LastSeenAt = sync.dbAccess.counter() + state.LastSeenAt = s.dbAccess.counter() state.Latest = storage.ZeroKey - log.Trace(fmt.Sprintf("syncer[%v]: sending %v", sync.key.Log(), unsynced)) - // send the unsynced keyssync + log.Trace(fmt.Sprintf("syncer[%v]: sending %v", s.key.Log(), unsynced)) + // send the unsynced keys stateCopy := *state - err := sync.unsyncedKeys(unsynced, &stateCopy) + err := s.unsyncedKeys(unsynced, &stateCopy) if err != nil { - log.Warn(fmt.Sprintf("syncer[%v]: unable to send unsynced keys: %v", sync.key.Log(), err)) + log.Warn(fmt.Sprintf("syncer[%v]: unable to send unsynced keys: %v", s.key.Log(), err)) } - sync.state = state - log.Debug(fmt.Sprintf("syncer[%v]: --> %v keys sent: (total: %v (%v), history: %v), sent sync state: %v", sync.key.Log(), len(unsynced), keyCounts, keyCount, historyCnt, stateCopy)) + s.state = state + log.Debug(fmt.Sprintf("syncer[%v]: --> %v keys sent: (total: %v (%v), history: %v), sent sync state: %v", s.key.Log(), len(unsynced), keyCounts, keyCount, historyCnt, stateCopy)) unsynced = nil keys = nil } // process item and add it to the batch select { - case <-sync.quit: + case <-s.quit: break LOOP case req, more = <-keys: if keys == history && !more { - log.Trace(fmt.Sprintf("syncer[%v]: syncing history segment complete", sync.key.Log())) + log.Trace(fmt.Sprintf("syncer[%v]: syncing history segment complete", s.key.Log())) // history channel is closed, waiting for new state (called from sync()) - syncStates = sync.syncStates + syncStates = s.syncStates state.Synced = true // this signals that the current segment is complete select { case state.synced <- false: - case <-sync.quit: + case <-s.quit: break LOOP } justSynced = true history = nil } case <-deliveryRequest: - log.Trace(fmt.Sprintf("syncer[%v]: peer ready to receive", sync.key.Log())) + log.Trace(fmt.Sprintf("syncer[%v]: peer ready to receive", s.key.Log())) // this 1 cap channel can wake up the loop // signaling that peer is ready to receive unsynced Keys @@ -518,23 +517,23 @@ LOOP: deliveryRequest = nil case <-newUnsyncedKeys: - log.Trace(fmt.Sprintf("syncer[%v]: new unsynced keys available", sync.key.Log())) + log.Trace(fmt.Sprintf("syncer[%v]: new unsynced keys available", s.key.Log())) // this 1 cap channel can wake up the loop // signals that data is available to send if peer is ready to receive newUnsyncedKeys = nil - keys = sync.keys[High] + keys = s.keys[High] case state, more = <-syncStates: // this resets the state if !more { - state = sync.state - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing complete upto %v)", sync.key.Log(), priority, state)) + state = s.state + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing complete upto %v)", s.key.Log(), priority, state)) state.Synced = true syncStates = nil } else { - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing history upto %v priority %v)", sync.key.Log(), priority, state, histPrior)) + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing history upto %v priority %v)", s.key.Log(), priority, state, histPrior)) state.Synced = false - history = sync.syncHistory(state) + history = s.syncHistory(state) // only one history at a time, only allow another one once the // history channel is closed syncStates = nil @@ -544,19 +543,19 @@ LOOP: continue LOOP } - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) added to unsynced keys: %v", sync.key.Log(), priority, req)) + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) added to unsynced keys: %v", s.key.Log(), priority, req)) keyCounts[priority]++ keyCount++ if keys == history { - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) history item %v (synced = %v)", sync.key.Log(), priority, req, state.Synced)) + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) history item %v (synced = %v)", s.key.Log(), priority, req, state.Synced)) historyCnt++ } - if sreq, err := sync.newSyncRequest(req, priority); err == nil { + if sreq, err := s.newSyncRequest(req, priority); err == nil { // extract key from req - log.Trace(fmt.Sprintf("syncer[%v]: (priority %v): request %v (synced = %v)", sync.key.Log(), priority, req, state.Synced)) + log.Trace(fmt.Sprintf("syncer[%v]: (priority %v): request %v (synced = %v)", s.key.Log(), priority, req, state.Synced)) unsynced = append(unsynced, sreq) } else { - log.Warn(fmt.Sprintf("syncer[%v]: (priority %v): error creating request for %v: %v)", sync.key.Log(), priority, req, err)) + log.Warn(fmt.Sprintf("syncer[%v]: (priority %v): error creating request for %v: %v)", s.key.Log(), priority, req, err)) } } @@ -565,7 +564,7 @@ LOOP: // delivery loop // takes into account priority, send store Requests with chunk (delivery) // idle blocking if no new deliveries in any of the queues -func (sync *syncer) syncDeliveries() { +func (s *syncer) syncDeliveries() { var req *storeRequestMsgData p := High var deliveries chan *storeRequestMsgData @@ -576,7 +575,7 @@ func (sync *syncer) syncDeliveries() { var total, success uint for { - deliveries = sync.deliveries[p] + deliveries = s.deliveries[p] select { case req = <-deliveries: n[p]++ @@ -585,13 +584,13 @@ func (sync *syncer) syncDeliveries() { if p == Low { // blocking, depletion on all channels, no preference for priority select { - case req = <-sync.deliveries[High]: + case req = <-s.deliveries[High]: n[High]++ - case req = <-sync.deliveries[Medium]: + case req = <-s.deliveries[Medium]: n[Medium]++ - case req = <-sync.deliveries[Low]: + case req = <-s.deliveries[Low]: n[Low]++ - case <-sync.quit: + case <-s.quit: return } p = High @@ -601,20 +600,20 @@ func (sync *syncer) syncDeliveries() { } } total++ - msg, err = sync.newStoreRequestMsgData(req) + msg, err = s.newStoreRequestMsgData(req) if err != nil { - log.Warn(fmt.Sprintf("syncer[%v]: failed to create store request for %v: %v", sync.key.Log(), req, err)) + log.Warn(fmt.Sprintf("syncer[%v]: failed to create store request for %v: %v", s.key.Log(), req, err)) } else { - err = sync.store(msg) + err = s.store(msg) if err != nil { - log.Warn(fmt.Sprintf("syncer[%v]: failed to deliver %v: %v", sync.key.Log(), req, err)) + log.Warn(fmt.Sprintf("syncer[%v]: failed to deliver %v: %v", s.key.Log(), req, err)) } else { success++ - log.Trace(fmt.Sprintf("syncer[%v]: %v successfully delivered", sync.key.Log(), req)) + log.Trace(fmt.Sprintf("syncer[%v]: %v successfully delivered", s.key.Log(), req)) } } - if total%sync.SyncBatchSize == 0 { - log.Debug(fmt.Sprintf("syncer[%v]: deliver Total: %v, Success: %v, High: %v/%v, Medium: %v/%v, Low %v/%v", sync.key.Log(), total, success, c[High], n[High], c[Medium], n[Medium], c[Low], n[Low])) + if total%s.SyncBatchSize == 0 { + log.Debug(fmt.Sprintf("syncer[%v]: deliver Total: %v, Success: %v, High: %v/%v, Medium: %v/%v, Low %v/%v", s.key.Log(), total, success, c[High], n[High], c[Medium], n[Medium], c[Low], n[Low])) } } } @@ -622,40 +621,37 @@ func (sync *syncer) syncDeliveries() { /* addRequest handles requests for delivery it accepts 4 types: - * storeRequestMsgData: coming from netstore propagate response * chunk: coming from forwarding (questionable: id?) * key: from incoming syncRequest * syncDbEntry: key,id encoded in db - If sync mode is on for the type of request, then it sends the request to the keys queue of the correct priority channel buffered with capacity (SyncBufferSize) - If sync mode is off then, requests are directly sent to deliveries */ -func (sync *syncer) addRequest(req interface{}, ty int) { +func (s *syncer) addRequest(req interface{}, ty int) { // retrieve priority for request type name int8 - priority := sync.SyncPriorities[ty] + priority := s.SyncPriorities[ty] // sync mode for this type ON - if sync.syncF() || ty == DeliverReq { - if sync.SyncModes[ty] { - sync.addKey(req, priority, sync.quit) + if s.syncF() || ty == DeliverReq { + if s.SyncModes[ty] { + s.addKey(req, priority, s.quit) } else { - sync.addDelivery(req, priority, sync.quit) + s.addDelivery(req, priority, s.quit) } } } // addKey queues sync request for sync confirmation with given priority // ie the key will go out in an unsyncedKeys message -func (sync *syncer) addKey(req interface{}, priority uint, quit chan bool) bool { +func (s *syncer) addKey(req interface{}, priority uint, quit chan bool) bool { select { - case sync.keys[priority] <- req: + case s.keys[priority] <- req: // this wakes up the unsynced keys loop if idle select { - case sync.newUnsyncedKeys <- true: + case s.newUnsyncedKeys <- true: default: } return true @@ -667,9 +663,9 @@ func (sync *syncer) addKey(req interface{}, priority uint, quit chan bool) bool // addDelivery queues delivery request for with given priority // ie the chunk will be delivered ASAP mod priority queueing handled by syncdb // requests are persisted across sessions for correct sync -func (sync *syncer) addDelivery(req interface{}, priority uint, quit chan bool) bool { +func (s *syncer) addDelivery(req interface{}, priority uint, quit chan bool) bool { select { - case sync.queues[priority].buffer <- req: + case s.queues[priority].buffer <- req: return true case <-quit: return false @@ -678,14 +674,14 @@ func (sync *syncer) addDelivery(req interface{}, priority uint, quit chan bool) // doDelivery delivers the chunk for the request with given priority // without queuing -func (sync *syncer) doDelivery(req interface{}, priority uint, quit chan bool) bool { - msgdata, err := sync.newStoreRequestMsgData(req) +func (s *syncer) doDelivery(req interface{}, priority uint, quit chan bool) bool { + msgdata, err := s.newStoreRequestMsgData(req) if err != nil { log.Warn(fmt.Sprintf("unable to deliver request %v: %v", msgdata, err)) return false } select { - case sync.deliveries[priority] <- msgdata: + case s.deliveries[priority] <- msgdata: return true case <-quit: return false @@ -694,9 +690,9 @@ func (sync *syncer) doDelivery(req interface{}, priority uint, quit chan bool) b // returns the delivery function for given priority // passed on to syncDb -func (sync *syncer) deliver(priority uint) func(req interface{}, quit chan bool) bool { +func (s *syncer) deliver(priority uint) func(req interface{}, quit chan bool) bool { return func(req interface{}, quit chan bool) bool { - return sync.doDelivery(req, priority, quit) + return s.doDelivery(req, priority, quit) } } @@ -704,23 +700,23 @@ func (sync *syncer) deliver(priority uint) func(req interface{}, quit chan bool) // depending on sync mode settings for BacklogReq, // re play of request db backlog sends items via confirmation // or directly delivers -func (sync *syncer) replay() func(req interface{}, quit chan bool) bool { - sync := sync.SyncModes[BacklogReq] - priority := sync.SyncPriorities[BacklogReq] +func (s *syncer) replay() func(req interface{}, quit chan bool) bool { + sync := s.SyncModes[BacklogReq] + priority := s.SyncPriorities[BacklogReq] // sync mode for this type ON if sync { return func(req interface{}, quit chan bool) bool { - return sync.addKey(req, priority, quit) + return s.addKey(req, priority, quit) } } return func(req interface{}, quit chan bool) bool { - return sync.doDelivery(req, priority, quit) + return s.doDelivery(req, priority, quit) } } // given a request, extends it to a full storeRequestMsgData // polimorphic: see addRequest for the types accepted -func (sync *syncer) newStoreRequestMsgData(req interface{}) (*storeRequestMsgData, error) { +func (s *syncer) newStoreRequestMsgData(req interface{}) (*storeRequestMsgData, error) { key, id, chunk, sreq, err := parseRequest(req) if err != nil { @@ -730,7 +726,7 @@ func (sync *syncer) newStoreRequestMsgData(req interface{}) (*storeRequestMsgDat if sreq == nil { if chunk == nil { var err error - chunk, err = sync.dbAccess.get(key) + chunk, err = s.dbAccess.get(key) if err != nil { return nil, err } @@ -758,7 +754,7 @@ func parseRequest(req interface{}) (storage.Key, uint64, *storage.Chunk, *storeR var err error if key, ok = req.(storage.Key); ok { - id = generateId() + id = generateID() } else if entry, ok = req.(*syncDbEntry); ok { id = binary.BigEndian.Uint64(entry.val[32:]) @@ -766,7 +762,7 @@ func parseRequest(req interface{}) (storage.Key, uint64, *storage.Chunk, *storeR } else if chunk, ok = req.(*storage.Chunk); ok { key = chunk.Key - id = generateId() + id = generateID() } else if sreq, ok = req.(*storeRequestMsgData); ok { key = sreq.Key