From c55b99418b04fa2d4eb88f4daeca1a945216afed Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Wed, 24 Jan 2018 17:56:36 +0100 Subject: [PATCH] Fix dbstore iterator bug and add even more logging --- p2p/protocols/protocol.go | 2 + swarm/network/stream/delivery.go | 26 +++++++------ swarm/network/stream/delivery_test.go | 4 +- swarm/network/stream/messages.go | 14 +++---- swarm/network/stream/stream.go | 2 +- swarm/network/stream/streamer_test.go | 4 +- swarm/network/stream/syncer.go | 8 ++-- swarm/network/stream/syncer_test.go | 5 +-- swarm/network/stream/testing/testing.go | 12 ++++-- swarm/storage/dbstore.go | 51 +++++++++++++------------ swarm/storage/localstore.go | 6 +-- 11 files changed, 71 insertions(+), 63 deletions(-) diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index 7b04069edf..48fc5e9fcc 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -34,6 +34,7 @@ import ( "reflect" "sync" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" ) @@ -210,6 +211,7 @@ func (p *Peer) Run(handler func(msg interface{}) error) error { // if they are useful for other protocols // overwrite Disconnect for testing, so that protocol readloop quits func (p *Peer) Drop(err error) { + log.Error("p2p protocol DROP", "err", err) p.Disconnect(p2p.DiscSubprotocolError) } diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 98d0a1631d..d397cd7b62 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -19,7 +19,6 @@ package stream import ( "errors" "fmt" - "os" "time" "github.com/ethereum/go-ethereum/log" @@ -100,9 +99,14 @@ func (s *SwarmChunkServer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64 } // GetData retrives chunk data from db store -func (s *SwarmChunkServer) GetData(key []byte) []byte { - chunk, _ := s.db.Get(storage.Key(key)) - return chunk.SData +func (s *SwarmChunkServer) GetData(key []byte) ([]byte, error) { + chunk, err := s.db.Get(storage.Key(key)) + if err == storage.ErrFetching { + <-chunk.ReqC + } else if err != nil { + return nil, err + } + return chunk.SData, nil } // RetrieveRequestMsg is the protocol msg for chunk retrieve requests @@ -141,7 +145,7 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e if req.SkipCheck { err := sp.Deliver(chunk, s.priority) if err != nil { - sp.Drop(err) + sp.Drop(fmt.Errorf("handleRetrieveRequestMsg: %v", err)) } } streamer.deliveryC <- chunk.Key[:] @@ -173,9 +177,9 @@ R: for req := range d.receiveC { // this should be has locally chunk, err := d.db.Get(req.Key) - fmt.Fprintln(os.Stderr, "pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) + log.Error("pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) if err == nil { - fmt.Fprintln(os.Stderr, "found existing?", "hash", chunk.Key.Hex()) + log.Error("found existing?", "hash", chunk.Key.Hex()) continue R } if err != storage.ErrFetching { @@ -183,19 +187,19 @@ R: } select { case <-chunk.ReqC: - fmt.Fprintln(os.Stderr, "someone else delivered?", "hash", chunk.Key.Hex()) + log.Error("someone else delivered?", "hash", chunk.Key.Hex()) continue R default: } go func() { chunk.SData = req.SData - fmt.Fprintln(os.Stderr, "received delivery", "hash", chunk.Key.Hex()) + log.Error("received delivery", "hash", chunk.Key.Hex()) d.db.Put(chunk) - fmt.Fprintln(os.Stderr, "put to db", "hash", chunk.Key.Hex()) + log.Error("put to db", "hash", chunk.Key.Hex()) chunk.WaitToStore() close(chunk.ReqC) //log.Warn("received delivery stored", "hash", chunk.Key) - fmt.Fprintln(os.Stderr, "requesters notified", "hash", chunk.Key.Hex()) + log.Error("requesters notified", "hash", chunk.Key.Hex()) d.counterDone++ }() } diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 904830535d..24b35ae122 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -380,7 +380,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck // which responds to chunk retrieve requests all but the last node in the chain does not var j int err := sim.CallClient(func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, errc, quitC) + err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC) if err != nil { return err } @@ -547,7 +547,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // which responds to chunk retrieve requests var j int simErrC <- sim.CallClient(func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, simErrC, quitC) + err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), simErrC, quitC) if err != nil { return err } diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 24a63391d8..a11c97332d 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -83,7 +83,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error { log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To) go func() { if err := p.SendOfferedHashes(os, req.From, req.To); err != nil { - p.Drop(err) + p.Drop(fmt.Errorf("handleSubscribeMsg SendOfferedHashes: %v", err)) } }() return nil @@ -176,7 +176,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { return case err := <-s.next: if err != nil { - p.Drop(err) + p.Drop(fmt.Errorf("handleOfferedHashesMsg next: %v", err)) return } case <-s.quit: @@ -185,7 +185,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To) err := p.SendPriority(msg, s.priority) if err != nil { - p.Drop(err) + p.Drop(fmt.Errorf("handleOfferedHashesMsg set priority: %v", err)) } }() return nil @@ -218,7 +218,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { // launch in go routine since GetBatch blocks until new hashes arrive go func() { if err := p.SendOfferedHashes(s, req.From, req.To); err != nil { - p.Drop(err) + p.Drop(fmt.Errorf("handleWantedHashesMsg SendOfferedHashes: %v", err)) } }() // go p.SendOfferedHashes(s, req.From, req.To) @@ -230,9 +230,9 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { for i := 0; i < l; i++ { if want.Get(i) { hash := hashes[i*HashSize : (i+1)*HashSize] - data := s.GetData(hash) - if data == nil { - return errors.New("not found") + data, err := s.GetData(hash) + if err != nil { + return fmt.Errorf("handleWantedHashesMsg get data %x: %v", hash, err) } chunk := storage.NewChunk(hash, nil) chunk.SData = data diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index a85745a539..ac6d027d2d 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -256,7 +256,7 @@ type server struct { // Server interface for outgoing peer Streamer type Server interface { SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) - GetData([]byte) []byte + GetData([]byte) ([]byte, error) } type client struct { diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index a905f4c963..fec23a06d3 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -81,8 +81,8 @@ func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, ui return make([]byte, HashSize), from + 1, to + 1, nil, nil } -func (self *testServer) GetData([]byte) []byte { - return nil +func (self *testServer) GetData([]byte) ([]byte, error) { + return nil, nil } func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) { diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index d85233e6df..af1bbb0b2e 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -73,14 +73,14 @@ func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) { } // GetSection retrieves the actual chunk from localstore -func (s *SwarmSyncerServer) GetData(key []byte) []byte { +func (s *SwarmSyncerServer) GetData(key []byte) ([]byte, error) { chunk, err := s.db.Get(storage.Key(key)) if err == storage.ErrFetching { <-chunk.ReqC } else if err != nil { - return nil + return nil, err } - return chunk.SData + return chunk.SData, nil } // GetBatch retrieves the next batch of hashes from the dbstore @@ -111,7 +111,7 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 } log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po)) - return batch, from, to + 1, nil, nil + return batch, from, to, nil, nil } // SwarmSyncerClient diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index dcfa81a3a5..263d230183 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -22,7 +22,6 @@ import ( "fmt" "io" "math" - "os" "testing" "time" @@ -142,7 +141,7 @@ 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) + err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC) if err != nil { return err } @@ -182,7 +181,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } else if err == nil { nodeHashFound++ } else { - fmt.Fprintln(os.Stderr, time.Now(), "not found", "index", i, "origin", j, "key", key.Hex(), "err", err) + log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) } } } diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index e1c50919e7..9585de369b 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -154,9 +154,9 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) { // set nodes number of Stores available stores, storeTeardown, err := SetStores(addrs...) teardown = func() { - storeTeardown() - adapterTeardown() net.Shutdown() + adapterTeardown() + storeTeardown() } if err != nil { return nil, teardown, err @@ -208,7 +208,7 @@ func (s *Simulation) Run(ctx context.Context, conf *RunConfig) (*simulations.Ste return result, nil } -func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) error { +func WatchDisconnections(id discover.NodeID, client *rpc.Client, expectedConnCount int, errc chan error, quitC chan struct{}) error { events := make(chan *p2p.PeerEvent) sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") if err != nil { @@ -218,10 +218,14 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error defer sub.Unsubscribe() select { case <-quitC: - return + if expectedConnCount <= 0 { + return + } case e := <-events: + expectedConnCount-- errc <- fmt.Errorf("peerEvent for node %v: %v", id, e) case err := <-sub.Err(): + expectedConnCount = 0 if err != nil { errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err) } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index 25de2dcb70..41b0b67a24 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -30,7 +30,6 @@ import ( "fmt" "io" "io/ioutil" - "os" "sync" "time" @@ -540,16 +539,16 @@ func (s *DbStore) CurrentStorageIndex() uint64 { } func (s *DbStore) Put(chunk *Chunk) { - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put", "hash", chunk.Key.Hex()) + log.Error("DbStore.Put", "hash", chunk.Key.Hex()) done := make(chan struct{}) defer close(done) go func() { - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex()) + log.Error("DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex()) select { case <-time.After(1 * time.Second): - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put WAITING", "hash", chunk.Key.Hex()) + log.Error("DbStore.Put WAITING", "hash", chunk.Key.Hex()) case <-done: - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put EXITED", "hash", chunk.Key.Hex()) + log.Error("DbStore.Put EXITED", "hash", chunk.Key.Hex()) } }() @@ -557,35 +556,35 @@ func (s *DbStore) Put(chunk *Chunk) { var index dpaDBIndex po := s.po(chunk.Key) - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get is being called...", "hash", chunk.Key.Hex()) + log.Error("DbStore.db.Get is being called...", "hash", chunk.Key.Hex()) - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquiring", "hash", chunk.Key.Hex()) + log.Error("DbStore.LOCK acquiring", "hash", chunk.Key.Hex()) s.lock.Lock() - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquired", "hash", chunk.Key.Hex()) + log.Error("DbStore.LOCK acquired", "hash", chunk.Key.Hex()) defer s.lock.Unlock() idata, err := s.db.Get(ikey) - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get done", "hash", chunk.Key.Hex(), "err", err) + log.Error("DbStore.db.Get done", "hash", chunk.Key.Hex(), "err", err) if err != nil { s.doPut(chunk, ikey, &index, po) + batchC := s.batchC go func() { defer func() { if err := recover(); err != nil { - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err) + log.Error("DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err) } }() + + <-batchC + close(chunk.dbStored) }() - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) + log.Error("DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) } else { log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access")) decodeIndex(idata, &index) - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put already found", "hash", chunk.Key.Hex()) - } - batchC := s.batchC - go func() { - <-batchC close(chunk.dbStored) - }() + log.Error("DbStore.Put already found", "hash", chunk.Key.Hex()) + } index.Access = s.accessCnt s.accessCnt++ idata = encodeIndex(&index) @@ -594,7 +593,7 @@ func (s *DbStore) Put(chunk *Chunk) { case s.batchesC <- struct{}{}: default: } - fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Put done", "hash", chunk.Key.Hex(), "err", err) + log.Error("DbStore.db.Put done", "hash", chunk.Key.Hex(), "err", err) } // force putting into db, does not check access index @@ -716,9 +715,9 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) { hash := hasher.Sum(nil) if !bytes.Equal(hash, key) { - log.Trace(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:])) + log.Error(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:])) s.delete(indx.Idx, getIndexKey(key), s.po(key)) - log.Warn("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'") + log.Error("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'") } } @@ -784,15 +783,18 @@ func (s *DbStore) Close() { // initialises a sync iterator from a syncToken (passed in with the handshake) func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, uint64) bool) error { - s.lock.Lock() - defer s.lock.Unlock() + // probably, the lock is not needed + // s.lock.Lock() + // defer s.lock.Unlock() + untilkey := getDataKey(until, po) it := s.db.NewIterator() seek := getDataKey(since, po) it.Seek(seek) defer it.Release() - for it.Valid() { + + for it.Next() { dbkey := it.Key() if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 { break @@ -803,9 +805,8 @@ func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, if !f(Key(key), binary.BigEndian.Uint64(dbkey[2:])) { break } - it.Next() } - return nil + return it.Error() } func databaseExists(path string) bool { diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 00dc10d17b..f7cc4092d6 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -19,9 +19,7 @@ package storage import ( "encoding/binary" "fmt" - "os" "path/filepath" - "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/swarm/storage/mock" @@ -98,9 +96,9 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error) func (self *LocalStore) Put(chunk *Chunk) { chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) self.memStore.Put(chunk) - fmt.Fprintln(os.Stderr, time.Now(), "put to memstore", "hash", chunk.Key.Hex()) + log.Error("put to memstore", "hash", chunk.Key.Hex()) self.DbStore.Put(chunk) - fmt.Fprintln(os.Stderr, time.Now(), "put to dbstore", "hash", chunk.Key.Hex()) + log.Error("put to dbstore", "hash", chunk.Key.Hex()) } // Get(chunk *Chunk) looks up a chunk in the local stores