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

temporary debug

Browse source
This commit is contained in:
zelig 2018-01-23 11:40:40 +01:00
parent 295512f5a8
commit 2aa9791fc7
5 changed files with 347 additions and 282 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
swarm/network/stream/messages.go
View file

@ -20,6 +20,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"sync" "sync"
"time"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
bv "github.com/ethereum/go-ethereum/swarm/network/bitvector" bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
@ -130,6 +131,19 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
}(wait) }(wait)
} }
} }
// done := make(chan bool)
// go func() {
// wg.Wait()
// close(done)
// }()
// go func() {
// select {
// case <-done:
// s.next <- s.batchDone(p, req, hashes)
// case <-time.After(1 * time.Second):
// p.Drop(errors.New("timeout waiting for batch to be delivered"))
// }
// }()
go func() { go func() {
wg.Wait() wg.Wait()
s.next <- s.batchDone(p, req, hashes) s.next <- s.batchDone(p, req, hashes)
@ -154,6 +168,9 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
} }
go func() { go func() {
select { select {
case <-time.After(1 * time.Second):
p.Drop(errors.New("timeout waiting for batch to be delivered"))
return
case err := <-s.next: case err := <-s.next:
if err != nil { if err != nil {
p.Drop(err) p.Drop(err)
@ -196,7 +213,12 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
} }
hashes := s.currentBatch hashes := s.currentBatch
// launch in go routine since GetBatch blocks until new hashes arrive // launch in go routine since GetBatch blocks until new hashes arrive
go p.SendOfferedHashes(s, req.From, req.To) go func() {
if err := p.SendOfferedHashes(s, req.From, req.To); err != nil {
p.Drop(err)
}
}()
// go p.SendOfferedHashes(s, req.From, req.To)
l := len(hashes) / HashSize l := len(hashes) / HashSize
want, err := bv.NewFromBytes(req.Want, l) want, err := bv.NewFromBytes(req.Want, l)
if err != nil { if err != nil {

3
swarm/network/stream/syncer.go
View file

@ -182,15 +182,14 @@ func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) {
// NeedData // NeedData
func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) { func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
chunk, _ := s.db.GetOrCreateRequest(key) chunk, _ := s.db.GetOrCreateRequest(key)
log.Warn("created request", "key", chunk.Key)
// TODO: we may want to request from this peer anyway even if the request exists // TODO: we may want to request from this peer anyway even if the request exists
if chunk.ReqC == nil { if chunk.ReqC == nil {
log.Error("oops this is found")
return nil return nil
} }
// create request and wait until the chunk data arrives and is stored // create request and wait until the chunk data arrives and is stored
return func() { return func() {
chunk.WaitToStore() chunk.WaitToStore()
log.Warn("stored", "key", chunk.Key)
} }
} }

30
swarm/network/stream/syncer_test.go
View file

@ -37,12 +37,12 @@ import (
const dataChunkCount = 500 const dataChunkCount = 500
func TestSyncerSimulation(t *testing.T) { func TestSyncerSimulation(t *testing.T) {
testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1) // testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1)
// testSyncBetweenNodes(t, 2, 1, dataChunkCount, false, 1) // testSyncBetweenNodes(t, 2, 1, dataChunkCount, false, 1)
testSyncBetweenNodes(t, 4, 1, dataChunkCount, true, 1) // testSyncBetweenNodes(t, 4, 1, dataChunkCount, true, 1)
// testSyncBetweenNodes(t, 4, 1, dataChunkCount, false, 1) // // testSyncBetweenNodes(t, 4, 1, dataChunkCount, false, 1)
testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1) // testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1)
// testSyncBetweenNodes(t, 8, 1, dataChunkCount, false, 1) // // testSyncBetweenNodes(t, 8, 1, dataChunkCount, false, 1)
testSyncBetweenNodes(t, 16, 1, dataChunkCount, true, 1) testSyncBetweenNodes(t, 16, 1, dataChunkCount, true, 1)
// testSyncBetweenNodes(t, 16, 1, dataChunkCount, false, 1) // testSyncBetweenNodes(t, 16, 1, dataChunkCount, false, 1)
} }
@ -103,7 +103,9 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
}) })
} }
errc := make(chan error, 1)
waitPeerErrC = make(chan error) waitPeerErrC = make(chan error)
quitC := make(chan struct{})
action := func(ctx context.Context) error { action := func(ctx context.Context) error {
// need to wait till an aynchronous process registers the peers in streamer.peers // need to wait till an aynchronous process registers the peers in streamer.peers
// that is used by Subscribe // that is used by Subscribe
@ -122,6 +124,10 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
// each node Subscribes to each other's swarmChunkServerStreamName // each node Subscribes to each other's swarmChunkServerStreamName
j := 0 j := 0
return sim.CallClient(func(client *rpc.Client) error { return sim.CallClient(func(client *rpc.Client) error {
err := streamTesting.WatchDisconnections(sim.IDs[j], client, errc, quitC)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(ctx, 1*time.Second) ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
defer cancel() defer cancel()
j++ j++
@ -135,6 +141,8 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
defer func() { checkC <- struct{}{} }() defer func() { checkC <- struct{}{} }()
select { select {
case err := <-errc:
return false, err
case <-ctx.Done(): case <-ctx.Done():
return false, ctx.Err() return false, ctx.Err()
default: default:
@ -148,13 +156,19 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
found++ found++
} }
} }
log.Debug("sync check", "bin", po, "found", found, "total", total) log.Error("sync check", "bin", po, "found", found, "total", total)
return found == total, nil pass := found == total
if !pass {
return false, nil
}
close(quitC)
return true, nil
} }
conf.Step = &simulations.Step{ conf.Step = &simulations.Step{
Action: action, Action: action,
Trigger: streamTesting.PivotTrigger(10*time.Millisecond, checkC, sim.IDs[0]), Trigger: streamTesting.PivotTrigger(100*time.Millisecond, checkC, sim.IDs[0]),
Expect: &simulations.Expectation{ Expect: &simulations.Expectation{
Nodes: sim.IDs[0:1], Nodes: sim.IDs[0:1],
Check: check, Check: check,

4
swarm/storage/dbstore.go
View file

@ -601,7 +601,9 @@ func (s *DbStore) writeBatches() {
s.lock.Unlock() s.lock.Unlock()
err := s.writeBatch(b, e, d, a) err := s.writeBatch(b, e, d, a)
// TODO: set this error on the batch, then tell the chunk // TODO: set this error on the batch, then tell the chunk
log.Trace(fmt.Sprintf("DbStore: spawn batch write (%d chunks): %v", b.Len(), err)) if err != nil {
log.Error(fmt.Sprintf("DbStore: spawn batch write (%d chunks): %v", b.Len(), err))
}
close(c) close(c)
if e >= s.capacity { if e >= s.capacity {
log.Trace(fmt.Sprintf("DbStore: collecting garbage...(%d chunks)", e)) log.Trace(fmt.Sprintf("DbStore: collecting garbage...(%d chunks)", e))

568
swarm/storage/memstore.go
View file

@ -19,10 +19,7 @@
package storage package storage
import ( import (
"fmt"
"sync" "sync"
"github.com/ethereum/go-ethereum/log"
) )
const ( const (
@ -32,286 +29,317 @@ const (
defaultCacheCapacity = 5000 defaultCacheCapacity = 5000
) )
// type MemStore struct {
// memtree *memTree
// entryCnt, capacity uint // stored entries
// accessCnt uint64 // access counter; oldest is thrown away when full
// dbAccessCnt uint64
// dbStore *DbStore
// lock sync.Mutex
// }
//
// /*
// a hash prefix subtree containing subtrees or one storage entry (but never both)
//
// - access[0] stores the smallest (oldest) access count value in this subtree
// - if it contains more subtrees and its subtree count is at least 4, access[1:2]
// stores the smallest access count in the first and second halves of subtrees
// (so that access[0] = min(access[1], access[2])
// - likewise, if subtree count is at least 8,
// access[1] = min(access[3], access[4])
// access[2] = min(access[5], access[6])
// (access[] is a binary tree inside the multi-bit leveled hash tree)
// */
//
// func NewMemStore(d *DbStore, capacity uint) (m *MemStore) {
// m = &MemStore{}
// m.memtree = newMemTree(memTreeFLW, nil, 0)
// m.dbStore = d
// m.setCapacity(capacity)
// return
// }
//
// type memTree struct {
// subtree []*memTree
// parent *memTree
// parentIdx uint
//
// bits uint // log2(subtree count)
// width uint // subtree count
//
// entry *Chunk // if subtrees are present, entry should be nil
// lastDBaccess uint64
// access []uint64
// }
//
// func newMemTree(b uint, parent *memTree, pidx uint) (node *memTree) {
// node = new(memTree)
// node.bits = b
// node.width = 1 << b
// node.subtree = make([]*memTree, node.width)
// node.access = make([]uint64, node.width-1)
// node.parent = parent
// node.parentIdx = pidx
// if parent != nil {
// parent.subtree[pidx] = node
// }
//
// return node
// }
//
// func (node *memTree) updateAccess(a uint64) {
// aidx := uint(0)
// var aa uint64
// oa := node.access[0]
// for node.access[aidx] == oa {
// node.access[aidx] = a
// if aidx > 0 {
// aa = node.access[((aidx-1)^1)+1]
// aidx = (aidx - 1) >> 1
// } else {
// pidx := node.parentIdx
// node = node.parent
// if node == nil {
// return
// }
// nn := node.subtree[pidx^1]
// if nn != nil {
// aa = nn.access[0]
// } else {
// aa = 0
// }
// aidx = (node.width + pidx - 2) >> 1
// }
//
// if (aa != 0) && (aa < a) {
// a = aa
// }
// }
// }
//
// func (s *MemStore) setCapacity(c uint) {
// s.lock.Lock()
// defer s.lock.Unlock()
//
// for c < s.entryCnt {
// s.removeOldest()
// }
// s.capacity = c
// }
//
// // entry (not its copy) is going to be in MemStore
// func (s *MemStore) Put(entry *Chunk) {
// if s.capacity == 0 {
// return
// }
//
// s.lock.Lock()
// defer s.lock.Unlock()
//
// if s.entryCnt >= s.capacity {
// s.removeOldest()
// }
//
// s.accessCnt++
//
// node := s.memtree
// bitpos := uint(0)
// for node.entry == nil {
// l := entry.Key.bits(bitpos, node.bits)
// st := node.subtree[l]
// if st == nil {
// st = newMemTree(memTreeLW, node, l)
// bitpos += node.bits
// node = st
// break
// }
// bitpos += node.bits
// node = st
// }
//
// if node.entry != nil {
//
// if node.entry.Key.isEqual(entry.Key) {
// node.updateAccess(s.accessCnt)
// if entry.SData == nil {
// entry.Size = node.entry.Size
// entry.SData = node.entry.SData
// }
// if entry.ReqC == nil {
// entry.ReqC = node.entry.ReqC
// }
// entry.C = node.entry.C
// node.entry = entry
// return
// }
//
// for node.entry != nil {
//
// l := node.entry.Key.bits(bitpos, node.bits)
// st := node.subtree[l]
// if st == nil {
// st = newMemTree(memTreeLW, node, l)
// }
// st.entry = node.entry
// node.entry = nil
// st.updateAccess(node.access[0])
//
// l = entry.Key.bits(bitpos, node.bits)
// st = node.subtree[l]
// if st == nil {
// st = newMemTree(memTreeLW, node, l)
// }
// bitpos += node.bits
// node = st
//
// }
// }
//
// node.entry = entry
// node.lastDBaccess = s.dbAccessCnt
// node.updateAccess(s.accessCnt)
// s.entryCnt++
// }
//
// func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) {
// s.lock.Lock()
// defer s.lock.Unlock()
//
// node := s.memtree
// bitpos := uint(0)
// for node.entry == nil {
// l := hash.bits(bitpos, node.bits)
// st := node.subtree[l]
// if st == nil {
// return nil, ErrNotFound
// }
// bitpos += node.bits
// node = st
// }
//
// if node.entry.Key.isEqual(hash) {
// s.accessCnt++
// node.updateAccess(s.accessCnt)
// chunk = node.entry
// if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt {
// s.dbAccessCnt++
// node.lastDBaccess = s.dbAccessCnt
// if s.dbStore != nil {
// s.dbStore.updateAccessCnt(hash)
// }
// }
// } else {
// err = ErrNotFound
// }
//
// return
// }
//
// func (s *MemStore) removeOldest() {
// node := s.memtree
// log.Warn("purge memstore")
// for node.entry == nil {
//
// aidx := uint(0)
// av := node.access[aidx]
//
// for aidx < node.width/2-1 {
// if av == node.access[aidx*2+1] {
// node.access[aidx] = node.access[aidx*2+2]
// aidx = aidx*2 + 1
// } else if av == node.access[aidx*2+2] {
// node.access[aidx] = node.access[aidx*2+1]
// aidx = aidx*2 + 2
// } else {
// panic(nil)
// }
// }
// pidx := aidx*2 + 2 - node.width
// if (node.subtree[pidx] != nil) && (av == node.subtree[pidx].access[0]) {
// if node.subtree[pidx+1] != nil {
// node.access[aidx] = node.subtree[pidx+1].access[0]
// } else {
// node.access[aidx] = 0
// }
// } else if (node.subtree[pidx+1] != nil) && (av == node.subtree[pidx+1].access[0]) {
// if node.subtree[pidx] != nil {
// node.access[aidx] = node.subtree[pidx].access[0]
// } else {
// node.access[aidx] = 0
// }
// pidx++
// } else {
// panic(nil)
// }
//
// //fmt.Println(pidx)
// node = node.subtree[pidx]
//
// }
//
// log.Trace(fmt.Sprintf("Memstore Clean: Waiting for chunk %v to be saved", node.entry.Key.Log()))
// <-node.entry.dbStored
// log.Trace(fmt.Sprintf("Memstore Clean: Chunk %v saved to DBStore. Ready to clear from mem.", node.entry.Key.Log()))
//
// if node.entry.ReqC == nil {
// node.entry = nil
// s.entryCnt--
// } else {
// return
// }
//
// node.access[0] = 0
//
// //---
//
// aidx := uint(0)
// for {
// aa := node.access[aidx]
// if aidx > 0 {
// aidx = (aidx - 1) >> 1
// } else {
// pidx := node.parentIdx
// node = node.parent
// if node == nil {
// return
// }
// aidx = (node.width + pidx - 2) >> 1
// }
// if (aa != 0) && ((aa < node.access[aidx]) || (node.access[aidx] == 0)) {
// node.access[aidx] = aa
// }
// }
// }
type MemStore struct { type MemStore struct {
memtree *memTree m map[string]*Chunk
entryCnt, capacity uint // stored entries mu sync.RWMutex
accessCnt uint64 // access counter; oldest is thrown away when full
dbAccessCnt uint64
dbStore *DbStore
lock sync.Mutex
} }
/*
a hash prefix subtree containing subtrees or one storage entry (but never both)
- access[0] stores the smallest (oldest) access count value in this subtree
- if it contains more subtrees and its subtree count is at least 4, access[1:2]
stores the smallest access count in the first and second halves of subtrees
(so that access[0] = min(access[1], access[2])
- likewise, if subtree count is at least 8,
access[1] = min(access[3], access[4])
access[2] = min(access[5], access[6])
(access[] is a binary tree inside the multi-bit leveled hash tree)
*/
func NewMemStore(d *DbStore, capacity uint) (m *MemStore) { func NewMemStore(d *DbStore, capacity uint) (m *MemStore) {
m = &MemStore{} return &MemStore{
m.memtree = newMemTree(memTreeFLW, nil, 0) m: make(map[string]*Chunk),
m.dbStore = d
m.setCapacity(capacity)
return
}
type memTree struct {
subtree []*memTree
parent *memTree
parentIdx uint
bits uint // log2(subtree count)
width uint // subtree count
entry *Chunk // if subtrees are present, entry should be nil
lastDBaccess uint64
access []uint64
}
func newMemTree(b uint, parent *memTree, pidx uint) (node *memTree) {
node = new(memTree)
node.bits = b
node.width = 1 << b
node.subtree = make([]*memTree, node.width)
node.access = make([]uint64, node.width-1)
node.parent = parent
node.parentIdx = pidx
if parent != nil {
parent.subtree[pidx] = node
}
return node
}
func (node *memTree) updateAccess(a uint64) {
aidx := uint(0)
var aa uint64
oa := node.access[0]
for node.access[aidx] == oa {
node.access[aidx] = a
if aidx > 0 {
aa = node.access[((aidx-1)^1)+1]
aidx = (aidx - 1) >> 1
} else {
pidx := node.parentIdx
node = node.parent
if node == nil {
return
}
nn := node.subtree[pidx^1]
if nn != nil {
aa = nn.access[0]
} else {
aa = 0
}
aidx = (node.width + pidx - 2) >> 1
}
if (aa != 0) && (aa < a) {
a = aa
}
} }
} }
func (s *MemStore) setCapacity(c uint) { func (m *MemStore) Get(key Key) (*Chunk, error) {
s.lock.Lock() m.mu.RLock()
defer s.lock.Unlock() defer m.mu.RUnlock()
c, ok := m.m[string(key[:])]
for c < s.entryCnt { if !ok {
s.removeOldest() return nil, ErrNotFound
} }
s.capacity = c return c, nil
} }
// entry (not its copy) is going to be in MemStore func (m *MemStore) Put(c *Chunk) {
func (s *MemStore) Put(entry *Chunk) { m.mu.Lock()
if s.capacity == 0 { defer m.mu.Unlock()
return m.m[string(c.Key[:])] = c
}
s.lock.Lock()
defer s.lock.Unlock()
if s.entryCnt >= s.capacity {
s.removeOldest()
}
s.accessCnt++
node := s.memtree
bitpos := uint(0)
for node.entry == nil {
l := entry.Key.bits(bitpos, node.bits)
st := node.subtree[l]
if st == nil {
st = newMemTree(memTreeLW, node, l)
bitpos += node.bits
node = st
break
}
bitpos += node.bits
node = st
}
if node.entry != nil {
if node.entry.Key.isEqual(entry.Key) {
node.updateAccess(s.accessCnt)
if entry.SData == nil {
entry.Size = node.entry.Size
entry.SData = node.entry.SData
}
if entry.ReqC == nil {
entry.ReqC = node.entry.ReqC
}
entry.C = node.entry.C
node.entry = entry
return
}
for node.entry != nil {
l := node.entry.Key.bits(bitpos, node.bits)
st := node.subtree[l]
if st == nil {
st = newMemTree(memTreeLW, node, l)
}
st.entry = node.entry
node.entry = nil
st.updateAccess(node.access[0])
l = entry.Key.bits(bitpos, node.bits)
st = node.subtree[l]
if st == nil {
st = newMemTree(memTreeLW, node, l)
}
bitpos += node.bits
node = st
}
}
node.entry = entry
node.lastDBaccess = s.dbAccessCnt
node.updateAccess(s.accessCnt)
s.entryCnt++
} }
func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) { func (m *MemStore) setCapacity(n int) {
s.lock.Lock()
defer s.lock.Unlock()
node := s.memtree
bitpos := uint(0)
for node.entry == nil {
l := hash.bits(bitpos, node.bits)
st := node.subtree[l]
if st == nil {
return nil, ErrNotFound
}
bitpos += node.bits
node = st
}
if node.entry.Key.isEqual(hash) {
s.accessCnt++
node.updateAccess(s.accessCnt)
chunk = node.entry
if s.dbAccessCnt-node.lastDBaccess > dbForceUpdateAccessCnt {
s.dbAccessCnt++
node.lastDBaccess = s.dbAccessCnt
if s.dbStore != nil {
s.dbStore.updateAccessCnt(hash)
}
}
} else {
err = ErrNotFound
}
return
}
func (s *MemStore) removeOldest() {
node := s.memtree
log.Warn("purge memstore")
for node.entry == nil {
aidx := uint(0)
av := node.access[aidx]
for aidx < node.width/2-1 {
if av == node.access[aidx*2+1] {
node.access[aidx] = node.access[aidx*2+2]
aidx = aidx*2 + 1
} else if av == node.access[aidx*2+2] {
node.access[aidx] = node.access[aidx*2+1]
aidx = aidx*2 + 2
} else {
panic(nil)
}
}
pidx := aidx*2 + 2 - node.width
if (node.subtree[pidx] != nil) && (av == node.subtree[pidx].access[0]) {
if node.subtree[pidx+1] != nil {
node.access[aidx] = node.subtree[pidx+1].access[0]
} else {
node.access[aidx] = 0
}
} else if (node.subtree[pidx+1] != nil) && (av == node.subtree[pidx+1].access[0]) {
if node.subtree[pidx] != nil {
node.access[aidx] = node.subtree[pidx].access[0]
} else {
node.access[aidx] = 0
}
pidx++
} else {
panic(nil)
}
//fmt.Println(pidx)
node = node.subtree[pidx]
}
log.Trace(fmt.Sprintf("Memstore Clean: Waiting for chunk %v to be saved", node.entry.Key.Log()))
<-node.entry.dbStored
log.Trace(fmt.Sprintf("Memstore Clean: Chunk %v saved to DBStore. Ready to clear from mem.", node.entry.Key.Log()))
if node.entry.ReqC == nil {
node.entry = nil
s.entryCnt--
} else {
return
}
node.access[0] = 0
//---
aidx := uint(0)
for {
aa := node.access[aidx]
if aidx > 0 {
aidx = (aidx - 1) >> 1
} else {
pidx := node.parentIdx
node = node.parent
if node == nil {
return
}
aidx = (node.width + pidx - 2) >> 1
}
if (aa != 0) && ((aa < node.access[aidx]) || (node.access[aidx] == 0)) {
node.access[aidx] = aa
}
}
} }
// Close memstore // Close memstore