forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-18 18:02:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/*: fixed naming capitalization errors

Browse source
This commit is contained in:
Eli 2018-05-02 15:44:20 -07:00
parent d02a4c5fee
commit 527c74902f
6 changed files with 259 additions and 273 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
swarm/fuse/fuse_file.go
View file

@ -82,7 +82,7 @@ func (file *SwarmFile) Attr(ctx context.Context, a *fuse.Attr) error {
a.Gid = uint32(os.Getegid()) a.Gid = uint32(os.Getegid())
if file.fileSize == -1 { if file.fileSize == -1 {
reader := file.mountInfo.swarmApi.Retrieve(file.key) reader := file.mountInfo.swarmAPI.Retrieve(file.key)
quitC := make(chan bool) quitC := make(chan bool)
size, err := reader.Size(quitC) size, err := reader.Size(quitC)
if err != nil { if err != nil {
@ -99,7 +99,7 @@ func (file *SwarmFile) Read(ctx context.Context, req *fuse.ReadRequest, resp *fu
file.lock.RLock() file.lock.RLock()
defer file.lock.RUnlock() defer file.lock.RUnlock()
if file.reader == nil { 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) buf := make([]byte, req.Size)
n, err := file.reader.ReadAt(buf, req.Offset) n, err := file.reader.ReadAt(buf, req.Offset)

6
swarm/fuse/swarmfs_util.go
View file

@ -47,7 +47,7 @@ func externalUnmount(mountPoint string) error {
} }
func addFileToSwarm(sf *SwarmFile, content []byte, size int) 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 { if err != nil {
return err return err
} }
@ -66,7 +66,7 @@ func addFileToSwarm(sf *SwarmFile, content []byte, size int) error {
} }
func removeFileFromSwarm(sf *SwarmFile) 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 { if err != nil {
return err return err
} }
@ -102,7 +102,7 @@ func removeDirectoryFromSwarm(sd *SwarmDir) error {
} }
func appendToExistingFileInSwarm(sf *SwarmFile, content []byte, offset int64, length int64) 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 { if err != nil {
return err return err
} }

182
swarm/network/kademlia/kademlia.go
View file

@ -109,31 +109,31 @@ func New(addr Address, params *KadParams) *Kademlia {
} }
// accessor for KAD base address // accessor for KAD base address
func (k *Kademlia) Addr() Address { func (kad *Kademlia) Addr() Address {
return k.addr return kad.addr
} }
// accessor for KAD active node count // accessor for KAD active node count
func (k *Kademlia) Count() int { func (kad *Kademlia) Count() int {
defer k.lock.Unlock() defer kad.lock.Unlock()
k.lock.Lock() kad.lock.Lock()
return k.count return kad.count
} }
// accessor for KAD active node count // accessor for KAD active node count
func (k *Kademlia) DBCount() int { func (kad *Kademlia) DBCount() int {
return k.db.count() return kad.db.count()
} }
// On is the entry point called when a new nodes is added // 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) // 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) { func (kad *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error) {
log.Debug(fmt.Sprintf("%v", k)) log.Debug(fmt.Sprintf("%v", kad))
defer k.lock.Unlock() defer kad.lock.Unlock()
k.lock.Lock() kad.lock.Lock()
index := k.proximityBin(node.Addr()) index := kad.proximityBin(node.Addr())
record := k.db.findOrCreate(index, node.Addr(), node.Url()) record := kad.db.findOrCreate(index, node.Addr(), node.Url())
if cb != nil { if cb != nil {
err = cb(record, node) 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 // insert in kademlia table of active nodes
bucket := k.buckets[index] bucket := kad.buckets[index]
// if bucket is full insertion replaces the worst node // if bucket is full insertion replaces the worst node
// TODO: give priority to peers with active traffic // TODO: give priority to peers with active traffic
if len(bucket) < k.BucketSize { // >= allows us to add peers beyond the bucketsize limitation if len(bucket) < kad.BucketSize { // >= allows us to add peers beyond the bucketsize limitation
k.buckets[index] = append(bucket, node) kad.buckets[index] = append(bucket, node)
bucketAddIndexCount[index].Inc(1) bucketAddIndexCount[index].Inc(1)
log.Debug(fmt.Sprintf("add node %v to table", node)) log.Debug(fmt.Sprintf("add node %v to table", node))
k.setProxLimit(index, true) kad.setProxLimit(index, true)
record.node = node record.node = node
k.count++ kad.count++
return nil return nil
} }
// always rotate peers // always rotate peers
idle := k.MaxIdleInterval idle := kad.MaxIdleInterval
var pos int var pos int
var replaced Node var replaced Node
for i, p := range bucket { 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)) log.Debug(fmt.Sprintf("all peers wanted, PO%03d bucket full", index))
return fmt.Errorf("bucket full") 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() replaced.Drop()
// actually replace in the row. When off(node) is called, the peer is no longer in the row // actually replace in the row. When off(node) is called, the peer is no longer in the row
bucket[pos] = node bucket[pos] = node
// there is no change in bucket cardinalities so no prox limit adjustment is needed // there is no change in bucket cardinalities so no prox limit adjustment is needed
record.node = node record.node = node
k.count++ kad.count++
return nil return nil
} }
// Off is the called when a node is taken offline (from the protocol main loop exit) // 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) { func (kad *Kademlia) Off(node Node, cb func(*NodeRecord, Node)) (err error) {
k.lock.Lock() kad.lock.Lock()
defer k.lock.Unlock() defer kad.lock.Unlock()
index := k.proximityBin(node.Addr()) index := kad.proximityBin(node.Addr())
bucketRmIndexCount[index].Inc(1) bucketRmIndexCount[index].Inc(1)
bucket := k.buckets[index] bucket := kad.buckets[index]
for i := 0; i < len(bucket); i++ { for i := 0; i < len(bucket); i++ {
if node.Addr() == bucket[i].Addr() { if node.Addr() == bucket[i].Addr() {
k.buckets[index] = append(bucket[:i], bucket[(i+1):]...) kad.buckets[index] = append(bucket[:i], bucket[(i+1):]...)
k.setProxLimit(index, false) kad.setProxLimit(index, false)
break break
} }
} }
record := k.db.index[node.Addr()] record := kad.db.index[node.Addr()]
// callback on remove // callback on remove
if cb != nil { if cb != nil {
cb(record, record.node) cb(record, record.node)
} }
record.node = nil record.node = nil
k.count-- kad.count--
log.Debug(fmt.Sprintf("remove node %v from table, population now is %v", node, k.count)) log.Debug(fmt.Sprintf("remove node %v from table, population now is %v", node, kad.count))
return 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 // 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) // adjust Prox (proxLimit and proxSize after an insertion/removal of nodes)
// caller holds the lock // 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) // if the change is outside the core (PO lower)
// and the change does not leave a bucket empty then // and the change does not leave a bucket empty then
// no adjustment needed // no adjustment needed
if r < k.proxLimit && len(k.buckets[r]) > 0 { if r < kad.proxLimit && len(kad.buckets[r]) > 0 {
return return
} }
// if on=a node was added, then r must be within prox limit so increment cardinality // if on=a node was added, then r must be within prox limit so increment cardinality
if on { if on {
k.proxSize++ kad.proxSize++
curr := len(k.buckets[k.proxLimit]) curr := len(kad.buckets[kad.proxLimit])
// if now core is big enough without the furthest bucket, then contract // if now core is big enough without the furthest bucket, then contract
// this can result in more than one bucket change // this can result in more than one bucket change
for k.proxSize >= k.ProxBinSize+curr && curr > 0 { for kad.proxSize >= kad.ProxBinSize+curr && curr > 0 {
k.proxSize -= curr kad.proxSize -= curr
k.proxLimit++ kad.proxLimit++
curr = len(k.buckets[k.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 return
} }
// otherwise // otherwise
if r >= k.proxLimit { if r >= kad.proxLimit {
k.proxSize-- kad.proxSize--
} }
// expand core by lowering prox limit until hit zero or cover the empty bucket or reached target cardinality // 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) && for (kad.proxSize < kad.ProxBinSize || r < kad.proxLimit) &&
k.proxLimit > 0 { kad.proxLimit > 0 {
// //
k.proxLimit-- kad.proxLimit--
k.proxSize += len(k.buckets[k.proxLimit]) kad.proxSize += len(kad.buckets[kad.proxLimit])
log.Trace(fmt.Sprintf("proxbin expansion (size: %v, limit: %v, bin: %v)", k.proxSize, k.proxLimit, r)) 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 as the target. The most proximate bin will be the union of the bins between
proxLimit and MaxProx. proxLimit and MaxProx.
*/ */
func (k *Kademlia) FindClosest(target Address, max int) []Node { func (kad *Kademlia) FindClosest(target Address, max int) []Node {
k.lock.Lock() kad.lock.Lock()
defer k.lock.Unlock() defer kad.lock.Unlock()
r := nodesByDistance{ r := nodesByDistance{
target: target, target: target,
} }
po := k.proximityBin(target) po := kad.proximityBin(target)
index := po index := po
step := 1 step := 1
log.Trace(fmt.Sprintf("serving %v nodes at %v (PO%02d)", max, index, po)) 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 var n int
for index >= 0 { for index >= 0 {
// add entire bucket // add entire bucket
for _, p := range k.buckets[index] { for _, p := range kad.buckets[index] {
r.push(p, limit) r.push(p, limit)
n++ n++
} }
// terminate if index reached the bottom or enough peers > min // 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) { if n >= min && (step < 0 || max == 0) {
break break
} }
// reach top most non-empty PO bucket, turn around // reach top most non-empty PO bucket, turn around
if index == k.MaxProx { if index == kad.MaxProx {
index = po index = po
step = -1 step = -1
} }
@ -304,15 +304,15 @@ func (k *Kademlia) FindClosest(target Address, max int) []Node {
return r.nodes return r.nodes
} }
func (k *Kademlia) Suggest() (*NodeRecord, bool, int) { func (kad *Kademlia) Suggest() (*NodeRecord, bool, int) {
defer k.lock.RUnlock() defer kad.lock.RUnlock()
k.lock.RLock() kad.lock.RLock()
return k.db.findBest(k.BucketSize, func(i int) int { return len(k.buckets[i]) }) return kad.db.findBest(kad.BucketSize, func(i int) int { return len(kad.buckets[i]) })
} }
// Add node records to kaddb (persisted node record db) // Add node records to kaddb (persisted node record db)
func (k *Kademlia) Add(nrs []*NodeRecord) { func (kad *Kademlia) Add(nrs []*NodeRecord) {
k.db.add(nrs, k.proximityBin) kad.db.add(nrs, kad.proximityBin)
} }
// nodesByDistance is a list of nodes, ordered by distance to target. // 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. node from the other.
*/ */
func (k *Kademlia) proximityBin(other Address) (ret int) { func (kad *Kademlia) proximityBin(other Address) (ret int) {
ret = proximity(k.addr, other) ret = proximity(kad.addr, other)
if ret > k.MaxProx { if ret > kad.MaxProx {
ret = k.MaxProx ret = kad.MaxProx
} }
return return
} }
// provides keyrange for chunk db iteration // provides keyrange for chunk db iteration
func (k *Kademlia) KeyRange(other Address) (start, stop Address) { func (kad *Kademlia) KeyRange(other Address) (start, stop Address) {
defer k.lock.RUnlock() defer kad.lock.RUnlock()
k.lock.RLock() kad.lock.RLock()
return KeyRange(k.addr, other, k.proxLimit) return KeyRange(kad.addr, other, kad.proxLimit)
} }
// save persists kaddb on disk (written to file on path in json format. // save persists kaddb on disk (written to file on path in json format.
func (k *Kademlia) Save(path string, cb func(*NodeRecord, Node)) error { func (kad *Kademlia) Save(path string, cb func(*NodeRecord, Node)) error {
return k.db.save(path, cb) return kad.db.save(path, cb)
} }
// Load(path) loads the node record database (kaddb) from file on path. // 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) { func (kad *Kademlia) Load(path string, cb func(*NodeRecord, Node) error) (err error) {
return k.db.load(path, cb) return kad.db.load(path, cb)
} }
// kademlia table + kaddb table displayed with ascii // kademlia table + kaddb table displayed with ascii
func (k *Kademlia) String() string { func (kad *Kademlia) String() string {
defer k.lock.RUnlock() defer kad.lock.RUnlock()
k.lock.RLock() kad.lock.RLock()
defer k.db.lock.RUnlock() defer kad.db.lock.RUnlock()
k.db.lock.RLock() kad.db.lock.RLock()
var rows []string var rows []string
rows = append(rows, "=========================================================================") 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("%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", k.count, len(k.db.index), k.proxLimit, k.proxSize)) 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", k.MaxProx, k.ProxBinSize, k.BucketSize)) 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)) rows = append(rows, fmt.Sprintf("============ PROX LIMIT: %d ==========================================", i))
} }
row := []string{fmt.Sprintf("%03d", i), fmt.Sprintf("%2d", len(bucket))} row := []string{fmt.Sprintf("%03d", i), fmt.Sprintf("%2d", len(bucket))}
var k int var k int
c := k.db.cursors[i] c := kad.db.cursors[i]
for ; k < len(bucket); k++ { for ; k < len(bucket); k++ {
p := bucket[(c+k)%len(bucket)] p := bucket[(c+k)%len(bucket)]
row = append(row, p.Addr().String()[:6]) row = append(row, p.Addr().String()[:6])
@ -425,16 +425,16 @@ func (k *Kademlia) String() string {
for ; k < 4; k++ { for ; k < 4; k++ {
row = append(row, " ") 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]) row = append(row, p.Addr.String()[:6])
if j == 3 { if j == 3 {
break break
} }
} }
rows = append(rows, strings.Join(row, " ")) rows = append(rows, strings.Join(row, " "))
if i == k.MaxProx { if i == kad.MaxProx {
} }
} }
rows = append(rows, "=========================================================================") rows = append(rows, "=========================================================================")
@ -442,12 +442,12 @@ func (k *Kademlia) String() string {
} }
//We have to build up the array of counters for each index //We have to build up the array of counters for each index
func (k *Kademlia) initMetricsVariables() { func (kad *Kademlia) initMetricsVariables() {
//create the arrays //create the arrays
bucketAddIndexCount = make([]metrics.Counter, k.MaxProx+1) bucketAddIndexCount = make([]metrics.Counter, kad.MaxProx+1)
bucketRmIndexCount = make([]metrics.Counter, k.MaxProx+1) bucketRmIndexCount = make([]metrics.Counter, kad.MaxProx+1)
//at each index create a metrics counter //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) 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) bucketRmIndexCount[i] = metrics.NewRegisteredCounter(fmt.Sprintf("network.kademlia.bucket.rm.%d.index", i), nil)
} }

2
swarm/network/protocol.go
View file

@ -229,7 +229,7 @@ func (bzz *bzz) handle() error {
return fmt.Errorf("<- %v: Data too short (%v)", msg, n) return fmt.Errorf("<- %v: Data too short (%v)", msg, n)
} }
// last Active time is set only when receiving chunks // 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())) log.Trace(fmt.Sprintf("incoming store request: %s", req.String()))
// swap accounting is done within forwarding // swap accounting is done within forwarding
bzz.storage.HandleStoreRequestMsg(&req, &peer{bzz: bzz}) bzz.storage.HandleStoreRequestMsg(&req, &peer{bzz: bzz})

124
swarm/network/syncdb.go
View file

@ -31,13 +31,11 @@ const counterKeyPrefix = 0x01
/* /*
syncDb is a queueing service for outgoing deliveries. syncDb is a queueing service for outgoing deliveries.
One instance per priority queue for each peer One instance per priority queue for each peer
a syncDb instance maintains an in-memory buffer (of capacity bufferSize) a syncDb instance maintains an in-memory buffer (of capacity bufferSize)
once its in-memory buffer is full it switches to persisting in db once its in-memory buffer is full it switches to persisting in db
and dbRead iterator iterates through the items keeping their order 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 once the db read catches up (there is no more items in the db) then
it switches back to in-memory buffer. it switches back to in-memory buffer.
when syncdb is stopped all items in the buffer are saved to the db when syncdb is stopped all items in the buffer are saved to the db
*/ */
type syncDb struct { 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 bufferRead is a forever iterator loop that takes care of delivering
outgoing store requests reads from incoming buffer outgoing store requests reads from incoming buffer
its argument is the deliver function taking the item as first argument its argument is the deliver function taking the item as first argument
and a quit channel as second. and a quit channel as second.
Closing of this channel is supposed to abort all waiting for delivery Closing of this channel is supposed to abort all waiting for delivery
(typically network write) (typically network write)
The iteration switches between 2 modes, The iteration switches between 2 modes,
* buffer mode reads the in-memory buffer and delivers the items directly * 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 * db mode reads from the buffer and writes to the db, parallelly another
routine is started that reads from the db and delivers items routine is started that reads from the db and delivers items
If there is buffer contention in buffer mode (slow network, high upload volume) If there is buffer contention in buffer mode (slow network, high upload volume)
syncdb switches to db mode and starts dbRead syncdb switches to db mode and starts dbRead
Once db backlog is delivered, it reverts back to in-memory buffer Once db backlog is delivered, it reverts back to in-memory buffer
It is automatically started when syncdb is initialised. It is automatically started when syncdb is initialised.
It saves the buffer to db upon receiving quit signal. syncDb#stop() 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 buffer, db chan interface{} // channels representing the two read modes
var more bool var more bool
var req interface{} var req interface{}
@ -116,18 +109,18 @@ func (db *syncDb) bufferRead(deliver func(interface{}, chan bool) bool) {
var inBatch, inDb int var inBatch, inDb int
batch := new(leveldb.Batch) batch := new(leveldb.Batch)
var dbSize chan int var dbSize chan int
quit := db.quit quit := sdb.quit
counterValue := make([]byte, 8) counterValue := make([]byte, 8)
// counter is used for keeping the items in order, persisted to db // counter is used for keeping the items in order, persisted to db
// start counter where db was at, 0 if not found // 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 var counter uint64
if err == nil { if err == nil {
counter = binary.BigEndian.Uint64(data) 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 { } 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: LOOP:
@ -139,26 +132,26 @@ LOOP:
// deliver request : this is blocking on network write so // deliver request : this is blocking on network write so
// it is passed the quit channel as argument, so that it returns // 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 // 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 { 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 // received quit signal, save request currently waiting delivery
// by switching to db mode and closing the buffer // by switching to db mode and closing the buffer
buffer = nil buffer = nil
db = db.buffer db = sdb.buffer
close(db) close(db)
quit = nil // needs to block the quit case in select quit = nil // needs to block the quit case in select
break // break from select, this item will be written to the db break // break from select, this item will be written to the db
} }
db.total++ sdb.total++
log.Trace(fmt.Sprintf("syncDb[%v/%v] deliver (db/total): %v/%v", db.key.Log(), db.priority, db.dbTotal, db.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 // by the time deliver returns, there were new writes to the buffer
// if buffer contention is detected, switch to db mode which drains // if buffer contention is detected, switch to db mode which drains
// the buffer so no process will block on pushing store requests // the buffer so no process will block on pushing store requests
if len(buffer) == cap(buffer) { 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 buffer = nil
db = db.buffer db = sdb.buffer
} }
continue LOOP continue LOOP
@ -167,30 +160,30 @@ LOOP:
if !more { if !more {
// only if quit is called, saved all the buffer // only if quit is called, saved all the buffer
binary.BigEndian.PutUint64(counterValue, counter) binary.BigEndian.PutUint64(counterValue, counter)
batch.Put(db.counterKey, counterValue) // persist counter in batch batch.Put(sdb.counterKey, counterValue) // persist counter in batch
db.writeSyncBatch(batch) // save batch sdb.writeSyncBatch(batch) // save batch
log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save current batch to db", db.key.Log(), db.priority)) log.Trace(fmt.Sprintf("syncDb[%v/%v] quitting: save current batch to db", sdb.key.Log(), sdb.priority))
break LOOP break LOOP
} }
db.dbTotal++ sdb.dbTotal++
db.total++ sdb.total++
// otherwise break after select // otherwise break after select
case dbSize = <-db.batch: case dbSize = <-sdb.batch:
// explicit request for batch // explicit request for batch
if inBatch == 0 && quit != nil { if inBatch == 0 && quit != nil {
// there was no writes since the last batch so db depleted // there was no writes since the last batch so db depleted
// switch to buffer mode // 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 db = nil
buffer = db.buffer buffer = sdb.buffer
dbSize <- 0 // indicates to 'caller' that batch has been written dbSize <- 0 // indicates to 'caller' that batch has been written
inDb = 0 inDb = 0
continue LOOP continue LOOP
} }
binary.BigEndian.PutUint64(counterValue, counter) binary.BigEndian.PutUint64(counterValue, counter)
batch.Put(db.counterKey, counterValue) batch.Put(sdb.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)) log.Debug(fmt.Sprintf("syncDb[%v/%v] write batch %v/%v - %x - %x", sdb.key.Log(), sdb.priority, inBatch, counter, sdb.counterKey, counterValue))
batch = db.writeSyncBatch(batch) batch = sdb.writeSyncBatch(batch)
dbSize <- inBatch // indicates to 'caller' that batch has been written dbSize <- inBatch // indicates to 'caller' that batch has been written
inBatch = 0 inBatch = 0
continue LOOP continue LOOP
@ -198,45 +191,45 @@ LOOP:
// closing syncDb#quit channel is used to signal to all goroutines to quit // closing syncDb#quit channel is used to signal to all goroutines to quit
case <-quit: case <-quit:
// need to save backlog, so switch to db mode // need to save backlog, so switch to db mode
db = db.buffer db = sdb.buffer
buffer = nil buffer = nil
quit = 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) close(db)
continue LOOP continue LOOP
} }
// only get here if we put req into db // only get here if we put req into db
entry, err = db.newSyncDbEntry(req, counter) entry, err = sdb.newSyncDbEntry(req, counter)
if err != nil { 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 continue LOOP
} }
batch.Put(entry.key, entry.val) 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 // if just switched to db mode and not quitting, then launch dbRead
// in a parallel go routine to send deliveries from db // in a parallel go routine to send deliveries from db
if inDb == 0 && quit != nil { if inDb == 0 && quit != nil {
log.Trace(fmt.Sprintf("syncDb[%v/%v] start dbRead", db.key.Log(), db.priority)) log.Trace(fmt.Sprintf("syncDb[%v/%v] start dbRead", sdb.key.Log(), sdb.priority))
go db.dbRead(true, counter, deliver) go sdb.dbRead(true, counter, deliver)
} }
inDb++ inDb++
inBatch++ inBatch++
counter++ counter++
// need to save the batch if it gets too large (== dbBatchSize) // need to save the batch if it gets too large (== dbBatchSize)
if inBatch%int(db.dbBatchSize) == 0 { if inBatch%int(sdb.dbBatchSize) == 0 {
batch = db.writeSyncBatch(batch) 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)) log.Info(fmt.Sprintf("syncDb[%v:%v]: saved %v keys (saved counter at %v)", sdb.key.Log(), sdb.priority, inBatch, counter))
close(db.done) close(sdb.done)
} }
// writes the batch to the db and returns a new batch object // writes the batch to the db and returns a new batch object
func (db *syncDb) writeSyncBatch(batch *leveldb.Batch) *leveldb.Batch { func (sdb *syncDb) writeSyncBatch(batch *leveldb.Batch) *leveldb.Batch {
err := db.db.Write(batch) err := sdb.db.Write(batch)
if err != nil { 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 batch
} }
return new(leveldb.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 (???) this is mainly to prevent crashes due to network output buffer contention (???)
as well as to make syncronisation resilient to disconnects as well as to make syncronisation resilient to disconnects
the messages are supposed to be sent in the p2p priority queue. the messages are supposed to be sent in the p2p priority queue.
the request DB is shared between peers, but domains for each syncdb the request DB is shared between peers, but domains for each syncdb
are disjoint. dbkeys (42 bytes) are structured: are disjoint. dbkeys (42 bytes) are structured:
* 0: 0x00 (0x01 reserved for counter key) * 0: 0x00 (0x01 reserved for counter key)
* 1: priorities - priority (so that high priority can be replayed first) * 1: priorities - priority (so that high priority can be replayed first)
* 2-33: peers address * 2-33: peers address
* 34-41: syncdb counter to preserve order (this field is missing for the counter key) * 34-41: syncdb counter to preserve order (this field is missing for the counter key)
values (40 bytes) are: values (40 bytes) are:
* 0-31: key * 0-31: key
* 32-39: request id * 32-39: request id
dbRead needs a boolean to indicate if on first round all the historical dbRead needs a boolean to indicate if on first round all the historical
record is synced. Second argument to indicate current db counter record is synced. Second argument to indicate current db counter
The third is the function to apply 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) key := make([]byte, 42)
copy(key, db.start) copy(key, sdb.start)
binary.BigEndian.PutUint64(key[34:], counter) binary.BigEndian.PutUint64(key[34:], counter)
var batches, n, cnt, total int var batches, n, cnt, total int
var more bool 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 // so that loop is not blocking while delivering
// only relevant if cnt is large // only relevant if cnt is large
select { select {
case db.batch <- batchSizes: case sdb.batch <- batchSizes:
case <-db.quit: case <-sdb.quit:
return return
} }
// wait for the write to finish and get the item count in the next batch // 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 return
} }
} }
it = db.db.NewIterator() it = sdb.db.NewIterator()
it.Seek(key) it.Seek(key)
if !it.Valid() { if !it.Valid() {
copy(key, db.start) copy(key, sdb.start)
useBatches = true useBatches = true
continue continue
} }
del = new(leveldb.Batch) 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() { for n = 0; !useBatches || n < cnt; it.Next() {
copy(key, it.Key()) copy(key, it.Key())
if len(key) == 0 || key[0] != 0 { if len(key) == 0 || key[0] != 0 {
copy(key, db.start) copy(key, sdb.start)
useBatches = true useBatches = true
break break
} }
val := make([]byte, 40) val := make([]byte, 40)
copy(val, it.Value()) copy(val, it.Value())
entry = &syncDbEntry{key, val} 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)) // 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, db.quit) more = fun(entry, sdb.quit)
if !more { if !more {
// quit received when waiting to deliver entry, the entry will not be deleted // 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 break
} }
// since subsequent batches of the same db session are indexed incrementally // 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++ n++
total++ 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)) 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))
db.db.Write(del) // this could be async called only when db is idle sdb.db.Write(del) // this could be async called only when db is idle
it.Release() it.Release()
} }
} }
// //
func (db *syncDb) stop() { func (sdb *syncDb) stop() {
close(db.quit) close(sdb.quit)
<-db.done <-sdb.done
} }
// calculate a dbkey for the request, for the db to work // calculate a dbkey for the request, for the db to work
// see syncdb for db key structure // see syncdb for db key structure
// polimorphic: accepted types, see syncer#addRequest // 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 key storage.Key
var chunk *storage.Chunk var chunk *storage.Chunk
var id uint64 var id uint64
@ -359,10 +349,10 @@ func (db *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDb
var sreq *storeRequestMsgData var sreq *storeRequestMsgData
if key, ok = req.(storage.Key); ok { if key, ok = req.(storage.Key); ok {
id = generateId() id = generateID()
} else if chunk, ok = req.(*storage.Chunk); ok { } else if chunk, ok = req.(*storage.Chunk); ok {
key = chunk.Key key = chunk.Key
id = generateId() id = generateID()
} else if sreq, ok = req.(*storeRequestMsgData); ok { } else if sreq, ok = req.(*storeRequestMsgData); ok {
key = sreq.Key key = sreq.Key
id = sreq.Id id = sreq.Id
@ -377,7 +367,7 @@ func (db *syncDb) newSyncDbEntry(req interface{}, counter uint64) (entry *syncDb
dbval := make([]byte, 40) dbval := make([]byte, 40)
// encode key // encode key
copy(dbkey[:], db.start[:34]) // db peer copy(dbkey[:], sdb.start[:34]) // db peer
binary.BigEndian.PutUint64(dbkey[34:], counter) binary.BigEndian.PutUint64(dbkey[34:], counter)
// encode value // encode value
copy(dbval, key[:]) copy(dbval, key[:])

214
swarm/network/syncer.go
View file

@ -262,24 +262,23 @@ func decodeSync(meta *json.RawMessage) (*syncState, error) {
* if Last < LastSeenAt then all items in between then process all * if Last < LastSeenAt then all items in between then process all
backlog from upto last disconnect backlog from upto last disconnect
* if Last > 0 && * if Last > 0 &&
sync is called from the syncer constructor and is not supposed to be used externally sync is called from the syncer constructor and is not supposed to be used externally
*/ */
func (sync *syncer) sync() { func (s *syncer) sync() {
state := sync.state state := s.state
// sync finished // sync finished
defer close(sync.syncStates) defer close(s.syncStates)
// 0. first replay stale requests from request db // 0. first replay stale requests from request db
if state.SessionAt == 0 { 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 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-- { 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 // unless peer is synced sync unfinished history beginning on
if !state.Synced { if !state.Synced {
@ -288,9 +287,9 @@ func (sync *syncer) sync() {
if !storage.IsZeroKey(state.Latest) { if !storage.IsZeroKey(state.Latest) {
// 1. there is unfinished earlier sync // 1. there is unfinished earlier sync
state.Start = state.Latest 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 // blocks while the entire history upto state is synced
sync.syncState(state) s.syncState(state)
if state.Last < state.SessionAt { if state.Last < state.SessionAt {
state.First = state.Last + 1 state.First = state.Last + 1
} }
@ -300,8 +299,8 @@ func (sync *syncer) sync() {
// 2. sync up to last disconnect1 // 2. sync up to last disconnect1
if state.First < state.LastSeenAt { if state.First < state.LastSeenAt {
state.Last = 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)) log.Debug(fmt.Sprintf("syncer[%v]: start syncronising history upto last disconnect at %v: %v", s.key.Log(), state.LastSeenAt, state))
sync.syncState(state) s.syncState(state)
state.First = state.LastSeenAt state.First = state.LastSeenAt
} }
state.Latest = storage.ZeroKey state.Latest = storage.ZeroKey
@ -315,28 +314,28 @@ func (sync *syncer) sync() {
// if there have been new chunks since last session // if there have been new chunks since last session
if state.LastSeenAt < state.SessionAt { if state.LastSeenAt < state.SessionAt {
state.Last = 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 // 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 // wait till syncronised block uptil state is synced
func (sync *syncer) syncState(state *syncState) { func (s *syncer) syncState(state *syncState) {
sync.syncStates <- state s.syncStates <- state
select { select {
case <-state.synced: case <-state.synced:
case <-sync.quit: case <-s.quit:
} }
} }
// stop quits both request processor and saves the request cache to disk // stop quits both request processor and saves the request cache to disk
func (sync *syncer) stop() { func (s *syncer) stop() {
close(sync.quit) close(s.quit)
log.Trace(fmt.Sprintf("syncer[%v]: stop and save sync request db backlog", sync.key.Log())) log.Trace(fmt.Sprintf("syncer[%v]: stop and save sync request db backlog", s.key.Log()))
for _, db := range sync.queues { for _, db := range s.queues {
db.stop() db.stop()
} }
} }
@ -351,7 +350,7 @@ func (req *syncRequest) String() string {
return fmt.Sprintf("<Key: %v, Priority: %v>", req.Key.Log(), req.Priority) return fmt.Sprintf("<Key: %v, Priority: %v>", 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) key, _, _, _, err := parseRequest(req)
// TODO: if req has chunk, it should be put in a cache // TODO: if req has chunk, it should be put in a cache
// create // 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 // * read is on demand, blocking unless history channel is read
// * accepts sync requests (syncStates) to create new db iterator // * accepts sync requests (syncStates) to create new db iterator
// * closes the channel one iteration finishes // * 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 var n uint
history := make(chan interface{}) 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)) 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 := sync.dbAccess.iterator(state) it := s.dbAccess.iterator(state)
if it != nil { if it != nil {
go func() { go func() {
// signal end of the iteration ended // 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 // blocking until history channel is read from
case history <- key: case history <- key:
n++ 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 state.Latest = key
case <-sync.quit: case <-s.quit:
return 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 return history
} }
// triggers key syncronisation // triggers key syncronisation
func (sync *syncer) sendUnsyncedKeys() { func (s *syncer) sendUnsyncedKeys() {
select { select {
case sync.deliveryRequest <- true: case s.deliveryRequest <- true:
default: default:
} }
} }
@ -410,7 +409,7 @@ func (sync *syncer) sendUnsyncedKeys() {
// historical data is used so historical items are lower priority within // historical data is used so historical items are lower priority within
// their priority group. // their priority group.
// * Order of historical data is unspecified // * Order of historical data is unspecified
func (sync *syncer) syncUnsyncedKeys() { func (s *syncer) syncUnsyncedKeys() {
// send out new // send out new
var unsynced []*syncRequest var unsynced []*syncRequest
var more, justSynced bool var more, justSynced bool
@ -418,12 +417,12 @@ func (sync *syncer) syncUnsyncedKeys() {
var history chan interface{} var history chan interface{}
priority := High priority := High
keys := sync.keys[priority] keys := s.keys[priority]
var newUnsyncedKeys, deliveryRequest chan bool var newUnsyncedKeys, deliveryRequest chan bool
keyCounts := make([]int, priorities) keyCounts := make([]int, priorities)
histPrior := sync.SyncPriorities[HistoryReq] histPrior := s.SyncPriorities[HistoryReq]
syncStates := sync.syncStates syncStates := s.syncStates
state := sync.state state := s.state
LOOP: LOOP:
for { for {
@ -439,15 +438,15 @@ LOOP:
PRIORITIES: PRIORITIES:
for priority = High; priority >= 0; priority-- { for priority = High; priority >= 0; priority-- {
// the first priority channel that is non-empty will be assigned to keys // the first priority channel that is non-empty will be assigned to keys
if len(sync.keys[priority]) > 0 { if len(s.keys[priority]) > 0 {
log.Trace(fmt.Sprintf("syncer[%v]: reading request with priority %v", sync.key.Log(), priority)) log.Trace(fmt.Sprintf("syncer[%v]: reading request with priority %v", s.key.Log(), priority))
keys = sync.keys[priority] keys = s.keys[priority]
break PRIORITIES 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 the input queue is empty on this level, resort to history if there is any
if uint(priority) == histPrior && history != nil { 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 keys = history
break PRIORITIES break PRIORITIES
} }
@ -457,8 +456,8 @@ LOOP:
// if peer ready to receive but nothing to send // if peer ready to receive but nothing to send
if keys == nil && deliveryRequest == nil { if keys == nil && deliveryRequest == nil {
// if no items left and switch to waiting mode // if no items left and switch to waiting mode
log.Trace(fmt.Sprintf("syncer[%v]: buffers consumed. Waiting", sync.key.Log())) log.Trace(fmt.Sprintf("syncer[%v]: buffers consumed. Waiting", s.key.Log()))
newUnsyncedKeys = sync.newUnsyncedKeys newUnsyncedKeys = s.newUnsyncedKeys
} }
// send msg iff // send msg iff
@ -469,48 +468,48 @@ LOOP:
if deliveryRequest == nil && if deliveryRequest == nil &&
(justSynced || (justSynced ||
len(unsynced) > 0 && keys == nil || len(unsynced) > 0 && keys == nil ||
len(unsynced) == int(sync.SyncBatchSize)) { len(unsynced) == int(s.SyncBatchSize)) {
justSynced = false justSynced = false
// listen to requests // listen to requests
deliveryRequest = sync.deliveryRequest deliveryRequest = s.deliveryRequest
newUnsyncedKeys = nil // not care about data until next req comes in newUnsyncedKeys = nil // not care about data until next req comes in
// set sync to current counter // set sync to current counter
// (all nonhistorical outgoing traffic sheduled and persisted // (all nonhistorical outgoing traffic sheduled and persisted
state.LastSeenAt = sync.dbAccess.counter() state.LastSeenAt = s.dbAccess.counter()
state.Latest = storage.ZeroKey state.Latest = storage.ZeroKey
log.Trace(fmt.Sprintf("syncer[%v]: sending %v", sync.key.Log(), unsynced)) log.Trace(fmt.Sprintf("syncer[%v]: sending %v", s.key.Log(), unsynced))
// send the unsynced keyssync // send the unsynced keys
stateCopy := *state stateCopy := *state
err := sync.unsyncedKeys(unsynced, &stateCopy) err := s.unsyncedKeys(unsynced, &stateCopy)
if err != nil { 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 s.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)) 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 unsynced = nil
keys = nil keys = nil
} }
// process item and add it to the batch // process item and add it to the batch
select { select {
case <-sync.quit: case <-s.quit:
break LOOP break LOOP
case req, more = <-keys: case req, more = <-keys:
if keys == history && !more { 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()) // 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 state.Synced = true // this signals that the current segment is complete
select { select {
case state.synced <- false: case state.synced <- false:
case <-sync.quit: case <-s.quit:
break LOOP break LOOP
} }
justSynced = true justSynced = true
history = nil history = nil
} }
case <-deliveryRequest: 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 // this 1 cap channel can wake up the loop
// signaling that peer is ready to receive unsynced Keys // signaling that peer is ready to receive unsynced Keys
@ -518,23 +517,23 @@ LOOP:
deliveryRequest = nil deliveryRequest = nil
case <-newUnsyncedKeys: 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 // this 1 cap channel can wake up the loop
// signals that data is available to send if peer is ready to receive // signals that data is available to send if peer is ready to receive
newUnsyncedKeys = nil newUnsyncedKeys = nil
keys = sync.keys[High] keys = s.keys[High]
case state, more = <-syncStates: case state, more = <-syncStates:
// this resets the state // this resets the state
if !more { if !more {
state = sync.state state = s.state
log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing complete upto %v)", sync.key.Log(), priority, state)) log.Trace(fmt.Sprintf("syncer[%v]: (priority %v) syncing complete upto %v)", s.key.Log(), priority, state))
state.Synced = true state.Synced = true
syncStates = nil syncStates = nil
} else { } 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 state.Synced = false
history = sync.syncHistory(state) history = s.syncHistory(state)
// only one history at a time, only allow another one once the // only one history at a time, only allow another one once the
// history channel is closed // history channel is closed
syncStates = nil syncStates = nil
@ -544,19 +543,19 @@ LOOP:
continue 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]++ keyCounts[priority]++
keyCount++ keyCount++
if keys == history { 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++ historyCnt++
} }
if sreq, err := sync.newSyncRequest(req, priority); err == nil { if sreq, err := s.newSyncRequest(req, priority); err == nil {
// extract key from req // 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) unsynced = append(unsynced, sreq)
} else { } 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 // delivery loop
// takes into account priority, send store Requests with chunk (delivery) // takes into account priority, send store Requests with chunk (delivery)
// idle blocking if no new deliveries in any of the queues // idle blocking if no new deliveries in any of the queues
func (sync *syncer) syncDeliveries() { func (s *syncer) syncDeliveries() {
var req *storeRequestMsgData var req *storeRequestMsgData
p := High p := High
var deliveries chan *storeRequestMsgData var deliveries chan *storeRequestMsgData
@ -576,7 +575,7 @@ func (sync *syncer) syncDeliveries() {
var total, success uint var total, success uint
for { for {
deliveries = sync.deliveries[p] deliveries = s.deliveries[p]
select { select {
case req = <-deliveries: case req = <-deliveries:
n[p]++ n[p]++
@ -585,13 +584,13 @@ func (sync *syncer) syncDeliveries() {
if p == Low { if p == Low {
// blocking, depletion on all channels, no preference for priority // blocking, depletion on all channels, no preference for priority
select { select {
case req = <-sync.deliveries[High]: case req = <-s.deliveries[High]:
n[High]++ n[High]++
case req = <-sync.deliveries[Medium]: case req = <-s.deliveries[Medium]:
n[Medium]++ n[Medium]++
case req = <-sync.deliveries[Low]: case req = <-s.deliveries[Low]:
n[Low]++ n[Low]++
case <-sync.quit: case <-s.quit:
return return
} }
p = High p = High
@ -601,20 +600,20 @@ func (sync *syncer) syncDeliveries() {
} }
} }
total++ total++
msg, err = sync.newStoreRequestMsgData(req) msg, err = s.newStoreRequestMsgData(req)
if err != nil { 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 { } else {
err = sync.store(msg) err = s.store(msg)
if err != nil { 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 { } else {
success++ 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 { 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", sync.key.Log(), total, success, c[High], n[High], c[Medium], n[Medium], c[Low], n[Low])) 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 addRequest handles requests for delivery
it accepts 4 types: it accepts 4 types:
* storeRequestMsgData: coming from netstore propagate response * storeRequestMsgData: coming from netstore propagate response
* chunk: coming from forwarding (questionable: id?) * chunk: coming from forwarding (questionable: id?)
* key: from incoming syncRequest * key: from incoming syncRequest
* syncDbEntry: key,id encoded in db * syncDbEntry: key,id encoded in db
If sync mode is on for the type of request, then If sync mode is on for the type of request, then
it sends the request to the keys queue of the correct priority it sends the request to the keys queue of the correct priority
channel buffered with capacity (SyncBufferSize) channel buffered with capacity (SyncBufferSize)
If sync mode is off then, requests are directly sent to deliveries 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 // retrieve priority for request type name int8
priority := sync.SyncPriorities[ty] priority := s.SyncPriorities[ty]
// sync mode for this type ON // sync mode for this type ON
if sync.syncF() || ty == DeliverReq { if s.syncF() || ty == DeliverReq {
if sync.SyncModes[ty] { if s.SyncModes[ty] {
sync.addKey(req, priority, sync.quit) s.addKey(req, priority, s.quit)
} else { } else {
sync.addDelivery(req, priority, sync.quit) s.addDelivery(req, priority, s.quit)
} }
} }
} }
// addKey queues sync request for sync confirmation with given priority // addKey queues sync request for sync confirmation with given priority
// ie the key will go out in an unsyncedKeys message // 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 { select {
case sync.keys[priority] <- req: case s.keys[priority] <- req:
// this wakes up the unsynced keys loop if idle // this wakes up the unsynced keys loop if idle
select { select {
case sync.newUnsyncedKeys <- true: case s.newUnsyncedKeys <- true:
default: default:
} }
return true 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 // addDelivery queues delivery request for with given priority
// ie the chunk will be delivered ASAP mod priority queueing handled by syncdb // ie the chunk will be delivered ASAP mod priority queueing handled by syncdb
// requests are persisted across sessions for correct sync // 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 { select {
case sync.queues[priority].buffer <- req: case s.queues[priority].buffer <- req:
return true return true
case <-quit: case <-quit:
return false 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 // doDelivery delivers the chunk for the request with given priority
// without queuing // without queuing
func (sync *syncer) doDelivery(req interface{}, priority uint, quit chan bool) bool { func (s *syncer) doDelivery(req interface{}, priority uint, quit chan bool) bool {
msgdata, err := sync.newStoreRequestMsgData(req) msgdata, err := s.newStoreRequestMsgData(req)
if err != nil { if err != nil {
log.Warn(fmt.Sprintf("unable to deliver request %v: %v", msgdata, err)) log.Warn(fmt.Sprintf("unable to deliver request %v: %v", msgdata, err))
return false return false
} }
select { select {
case sync.deliveries[priority] <- msgdata: case s.deliveries[priority] <- msgdata:
return true return true
case <-quit: case <-quit:
return false 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 // returns the delivery function for given priority
// passed on to syncDb // 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 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, // depending on sync mode settings for BacklogReq,
// re play of request db backlog sends items via confirmation // re play of request db backlog sends items via confirmation
// or directly delivers // or directly delivers
func (sync *syncer) replay() func(req interface{}, quit chan bool) bool { func (s *syncer) replay() func(req interface{}, quit chan bool) bool {
sync := sync.SyncModes[BacklogReq] sync := s.SyncModes[BacklogReq]
priority := sync.SyncPriorities[BacklogReq] priority := s.SyncPriorities[BacklogReq]
// sync mode for this type ON // sync mode for this type ON
if sync { if sync {
return func(req interface{}, quit chan bool) bool { 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 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 // given a request, extends it to a full storeRequestMsgData
// polimorphic: see addRequest for the types accepted // 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) key, id, chunk, sreq, err := parseRequest(req)
if err != nil { if err != nil {
@ -730,7 +726,7 @@ func (sync *syncer) newStoreRequestMsgData(req interface{}) (*storeRequestMsgDat
if sreq == nil { if sreq == nil {
if chunk == nil { if chunk == nil {
var err error var err error
chunk, err = sync.dbAccess.get(key) chunk, err = s.dbAccess.get(key)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -758,7 +754,7 @@ func parseRequest(req interface{}) (storage.Key, uint64, *storage.Chunk, *storeR
var err error var err error
if key, ok = req.(storage.Key); ok { if key, ok = req.(storage.Key); ok {
id = generateId() id = generateID()
} else if entry, ok = req.(*syncDbEntry); ok { } else if entry, ok = req.(*syncDbEntry); ok {
id = binary.BigEndian.Uint64(entry.val[32:]) 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 { } else if chunk, ok = req.(*storage.Chunk); ok {
key = chunk.Key key = chunk.Key
id = generateId() id = generateID()
} else if sreq, ok = req.(*storeRequestMsgData); ok { } else if sreq, ok = req.(*storeRequestMsgData); ok {
key = sreq.Key key = sreq.Key