From 2aa9791fc77d3d9f28c9b54e636ce25f2cc8ae99 Mon Sep 17 00:00:00 2001 From: zelig Date: Tue, 23 Jan 2018 11:40:40 +0100 Subject: [PATCH] temporary debug --- swarm/network/stream/messages.go | 24 +- swarm/network/stream/syncer.go | 3 +- swarm/network/stream/syncer_test.go | 30 +- swarm/storage/dbstore.go | 4 +- swarm/storage/memstore.go | 568 +++++++++++++++------------- 5 files changed, 347 insertions(+), 282 deletions(-) diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index a575b915a6..1cf9aeab43 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -20,6 +20,7 @@ import ( "errors" "fmt" "sync" + "time" "github.com/ethereum/go-ethereum/log" bv "github.com/ethereum/go-ethereum/swarm/network/bitvector" @@ -130,6 +131,19 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { }(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() { wg.Wait() s.next <- s.batchDone(p, req, hashes) @@ -154,6 +168,9 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { } go func() { select { + case <-time.After(1 * time.Second): + p.Drop(errors.New("timeout waiting for batch to be delivered")) + return case err := <-s.next: if err != nil { p.Drop(err) @@ -196,7 +213,12 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { } hashes := s.currentBatch // 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 want, err := bv.NewFromBytes(req.Want, l) if err != nil { diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index 99436e7ecf..eff42b4cd0 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -182,15 +182,14 @@ func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) { // NeedData func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) { 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 if chunk.ReqC == nil { + log.Error("oops this is found") return nil } // create request and wait until the chunk data arrives and is stored return func() { chunk.WaitToStore() - log.Warn("stored", "key", chunk.Key) } } diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 11de0bf0c7..d57be133cb 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -37,12 +37,12 @@ import ( const dataChunkCount = 500 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, 4, 1, dataChunkCount, true, 1) - // testSyncBetweenNodes(t, 4, 1, dataChunkCount, false, 1) - testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1) - // testSyncBetweenNodes(t, 8, 1, dataChunkCount, false, 1) + // testSyncBetweenNodes(t, 4, 1, dataChunkCount, true, 1) + // // testSyncBetweenNodes(t, 4, 1, dataChunkCount, false, 1) + // testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1) + // // testSyncBetweenNodes(t, 8, 1, dataChunkCount, false, 1) testSyncBetweenNodes(t, 16, 1, dataChunkCount, true, 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) + quitC := make(chan struct{}) action := func(ctx context.Context) error { // need to wait till an aynchronous process registers the peers in streamer.peers // 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 j := 0 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) defer cancel() j++ @@ -135,6 +141,8 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck defer func() { checkC <- struct{}{} }() select { + case err := <-errc: + return false, err case <-ctx.Done(): return false, ctx.Err() default: @@ -148,13 +156,19 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck found++ } } - log.Debug("sync check", "bin", po, "found", found, "total", total) - return found == total, nil + log.Error("sync check", "bin", po, "found", found, "total", total) + pass := found == total + if !pass { + return false, nil + } + close(quitC) + return true, nil + } conf.Step = &simulations.Step{ 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{ Nodes: sim.IDs[0:1], Check: check, diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index 0facd67c04..d8dace4758 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -601,7 +601,9 @@ func (s *DbStore) writeBatches() { s.lock.Unlock() err := s.writeBatch(b, e, d, a) // 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) if e >= s.capacity { log.Trace(fmt.Sprintf("DbStore: collecting garbage...(%d chunks)", e)) diff --git a/swarm/storage/memstore.go b/swarm/storage/memstore.go index 65affc3ffe..31e2baf454 100644 --- a/swarm/storage/memstore.go +++ b/swarm/storage/memstore.go @@ -19,10 +19,7 @@ package storage import ( - "fmt" "sync" - - "github.com/ethereum/go-ethereum/log" ) const ( @@ -32,286 +29,317 @@ const ( 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 { - memtree *memTree - entryCnt, capacity uint // stored entries - accessCnt uint64 // access counter; oldest is thrown away when full - dbAccessCnt uint64 - dbStore *DbStore - lock sync.Mutex + m map[string]*Chunk + mu sync.RWMutex } -/* -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 - } + return &MemStore{ + m: make(map[string]*Chunk), } } -func (s *MemStore) setCapacity(c uint) { - s.lock.Lock() - defer s.lock.Unlock() - - for c < s.entryCnt { - s.removeOldest() +func (m *MemStore) Get(key Key) (*Chunk, error) { + m.mu.RLock() + defer m.mu.RUnlock() + c, ok := m.m[string(key[:])] + if !ok { + return nil, ErrNotFound } - s.capacity = c + return c, nil } -// 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 (m *MemStore) Put(c *Chunk) { + m.mu.Lock() + defer m.mu.Unlock() + m.m[string(c.Key[:])] = c } -func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) { - s.lock.Lock() - defer s.lock.Unlock() +func (m *MemStore) setCapacity(n int) { - 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