diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index d397cd7b62..5a5263f6a0 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -60,6 +60,7 @@ type SwarmChunkServer struct { batchC chan []byte db *storage.DBAPI currentLen uint64 + quit chan struct{} } // NewSwarmChunkServer is SwarmChunkServer constructor @@ -68,6 +69,7 @@ func NewSwarmChunkServer(db *storage.DBAPI) *SwarmChunkServer { deliveryC: make(chan []byte, deliveryCap), batchC: make(chan []byte), db: db, + quit: make(chan struct{}), } go s.processDeliveries() return s @@ -79,6 +81,8 @@ func (s *SwarmChunkServer) processDeliveries() { var batchC chan []byte for { select { + case <-s.quit: + return case hash := <-s.deliveryC: hashes = append(hashes, hash...) batchC = s.batchC @@ -98,6 +102,11 @@ func (s *SwarmChunkServer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64 return } +// Close needs to be called on a stream server +func (s *SwarmChunkServer) Close() { + close(s.quit) +} + // GetData retrives chunk data from db store func (s *SwarmChunkServer) GetData(key []byte) ([]byte, error) { chunk, err := s.db.Get(storage.Key(key)) @@ -168,30 +177,40 @@ type ChunkDeliveryMsg struct { func (d *Delivery) handleChunkDeliveryMsg(sp *Peer, req *ChunkDeliveryMsg) error { d.counterIn++ + log.Error("push to receiveC", "hash", storage.Key(req.Key).Hex()) d.receiveC <- req return nil } func (d *Delivery) processReceivedChunks() { -R: + done := make(chan struct{}) + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + // R: for req := range d.receiveC { - // this should be has locally - chunk, err := d.db.Get(req.Key) - log.Error("pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) - if err == nil { - log.Error("found existing?", "hash", chunk.Key.Hex()) - continue R - } - if err != storage.ErrFetching { - panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk)) - } - select { - case <-chunk.ReqC: - log.Error("someone else delivered?", "hash", chunk.Key.Hex()) - continue R - default: - } - go func() { + log.Error("pop from receiveC", "hash", storage.Key(req.Key).Hex()) + timer.Reset(1 * time.Second) + go func(req *ChunkDeliveryMsg) { + defer func() { done <- struct{}{} }() + // this should be has locally + chunk, err := d.db.Get(req.Key) + log.Error("after db.Get", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) + if err == nil { + log.Error("found existing?", "hash", chunk.Key.Hex()) + // continue R + return + } + if err != storage.ErrFetching { + panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk)) + } + select { + case <-chunk.ReqC: + log.Error("someone else delivered?", "hash", chunk.Key.Hex()) + // continue R + return + default: + } + // go func() { chunk.SData = req.SData log.Error("received delivery", "hash", chunk.Key.Hex()) d.db.Put(chunk) @@ -201,7 +220,13 @@ R: //log.Warn("received delivery stored", "hash", chunk.Key) log.Error("requesters notified", "hash", chunk.Key.Hex()) d.counterDone++ - }() + // }() + }(req) + select { + case <-timer.C: + log.Error("!!!unable to process", "hash", req.Key.Hex()) + case <-done: + } } } diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 24b35ae122..05bac692e0 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -378,22 +378,23 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck // each node subscribes to the upstream swarm chunk server stream // 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, peerCount(sim.IDs[j]), errc, quitC) + for j := 0; j < nodes-1; j++ { + id := sim.IDs[j] + err := sim.CallClient(id, func(client *rpc.Client) error { + err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + j++ + sid := sim.IDs[j] + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false) + }) if err != nil { return err } - ctx, cancel := context.WithTimeout(ctx, 1*time.Second) - defer cancel() - j++ - sid := sim.IDs[j] - return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false) - }, sim.IDs[0:nodes-1]...) - if err != nil { - return err } - // create a retriever dpa for the pivot node delivery := deliveries[sim.IDs[0]] retrieveFunc := func(chunk *storage.Chunk) error { @@ -426,22 +427,21 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck default: } var total int64 - err := sim.CallClient(func(client *rpc.Client) error { + err := sim.CallClient(id, func(client *rpc.Client) error { ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() return client.CallContext(ctx, &total, "stream_readAll", common.BytesToHash(fileHash)) - }, id) + }) log.Info(fmt.Sprintf("check if %08x is available locally: number of bytes read %v/%v (error: %v)", fileHash, total, size, err)) if err != nil || total != int64(size) { 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.Trigger(10*time.Millisecond, quitC, sim.IDs[0]), // we are only testing the pivot node (net.Nodes[0]) Expect: &simulations.Expectation{ Nodes: sim.IDs[0:1], @@ -490,7 +490,12 @@ func BenchmarkDeliveryFromNodesWithCheck(b *testing.B) { } func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skipCheck bool) { + defaultSkipCheck = skipCheck toAddr = network.NewAddrFromNodeID + timeout := 300 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + conf := &streamTesting.RunConfig{ Adapter: *adapter, NodeCount: nodes, @@ -498,12 +503,12 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip ToAddr: toAddr, Services: services, } - defaultSkipCheck = skipCheck sim, teardown, err := streamTesting.NewSimulation(conf) defer teardown() if err != nil { b.Fatal(err.Error()) } + stores = make(map[discover.NodeID]storage.ChunkStore) deliveries = make(map[discover.NodeID]*Delivery) for i, id := range sim.IDs { @@ -515,17 +520,19 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip } return 2 } + // wait channel for all nodes all peer connections to set up + waitPeerErrC = make(chan error) + // create a dpa for the last node in the chain which we are gonna write to remoteDpa := storage.NewDPA(sim.Stores[nodes-1], storage.NewChunkerParams()) remoteDpa.Start() defer remoteDpa.Stop() - // wait channel for all nodes all peer connections to set up - waitPeerErrC = make(chan error) // channel to signal simulation initialisation with action call complete // or node disconnections simErrC := make(chan error) quitC := make(chan struct{}) + defer close(quitC) action := func(ctx context.Context) error { // each node Subscribes to each other's swarmChunkServerStreamName @@ -545,18 +552,19 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // each node except the last one subscribes to the upstream swarm chunk server stream // which responds to chunk retrieve requests - var j int - simErrC <- sim.CallClient(func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), simErrC, quitC) - if err != nil { - return err - } - ctx, cancel := context.WithTimeout(ctx, 1*time.Second) - defer cancel() - j++ - sid := sim.IDs[j] // the upstream peer's id - return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false) - }, sim.IDs[0:nodes-1]...) + for j := 0; j < nodes-1; j++ { + id := sim.IDs[j] + simErrC <- sim.CallClient(id, func(client *rpc.Client) error { + err := streamTesting.WatchDisconnections(id, client, peerCount(id), simErrC, quitC) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + sid := sim.IDs[j+1] // the upstream peer's id + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false) + }) + } // signal to the benchmark that setup is complete return err } @@ -589,10 +597,6 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip // run the simulation in the background errc := make(chan error) go func() { - timeout := 300 * time.Second - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - _, err := sim.Run(ctx, conf) errc <- err }() @@ -608,6 +612,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip select { case err = <-simErrC: case <-quitC: + return } trigger <- sim.IDs[0] checkC <- err @@ -669,7 +674,6 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip } } // benchmark over, trigger the check function to conclude the simulation - close(quitC) err = <-errc if err != nil { b.Fatalf("expected no error. got %v", err) diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index a11c97332d..aa8f5f75c4 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -179,8 +179,6 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { p.Drop(fmt.Errorf("handleOfferedHashesMsg next: %v", err)) return } - case <-s.quit: - return } 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) diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index ed2366f5c6..cd7cdc2c9a 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -83,6 +83,10 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { if err != nil { return err } + // true only when quiting + if len(hashes) == 0 { + return nil + } if proof == nil { proof = &HandoverProof{ Handover: &Handover{}, @@ -171,3 +175,9 @@ func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bo next <- nil // this is to allow wantedKeysMsg before first batch arrives return nil } + +func (p *Peer) close() { + for _, s := range p.servers { + s.Close() + } +} diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index ac6d027d2d..30cd71548f 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -198,7 +198,8 @@ func (r *Registry) run(p *protocols.Peer) error { sp := NewPeer(p, r) r.setPeer(sp) defer r.deletePeer(sp) - defer close(sp.quit) + // defer close(sp.quit + defer sp.close() return sp.Run(sp.HandleMsg) } @@ -257,6 +258,7 @@ type server struct { type Server interface { SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) GetData([]byte) ([]byte, error) + Close() } type client struct { @@ -266,8 +268,8 @@ type client struct { live bool stream string key []byte - quit chan struct{} - next chan error + // quit chan struct{} + next chan error } // Client interface for incoming peer Streamer diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index fec23a06d3..71c0b4bda9 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -85,6 +85,9 @@ func (self *testServer) GetData([]byte) ([]byte, error) { return nil, nil } +func (self *testServer) Close() { +} + func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(t) defer teardown() diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index af1bbb0b2e..9620a625da 100644 --- a/swarm/network/stream/syncer.go +++ b/swarm/network/stream/syncer.go @@ -42,6 +42,7 @@ type SwarmSyncerServer struct { db *storage.DBAPI sessionAt uint64 start uint64 + quit chan struct{} } // NewSwarmSyncerServer is contructor for SwarmSyncerServer @@ -56,6 +57,7 @@ func NewSwarmSyncerServer(live bool, po uint8, db *storage.DBAPI) (*SwarmSyncerS db: db, sessionAt: sessionAt, start: start, + quit: make(chan struct{}), }, nil } @@ -72,6 +74,11 @@ func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) { // }) } +// Close needs to be called on a stream server +func (s *SwarmSyncerServer) Close() { + close(s.quit) +} + // GetSection retrieves the actual chunk from localstore func (s *SwarmSyncerServer) GetData(key []byte) ([]byte, error) { chunk, err := s.db.Get(storage.Key(key)) @@ -95,7 +102,12 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6 } ticker := time.NewTicker(10 * time.Millisecond) defer ticker.Stop() - for range ticker.C { + for { + select { + case <-ticker.C: + case <-s.quit: + return nil, 0, 0, nil, nil + } err := s.db.Iterator(from, to, s.po, func(key storage.Key, idx uint64) bool { batch = append(batch, key[:]...) i++ diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 263d230183..1bcd5fe1aa 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -44,6 +44,7 @@ func TestSyncerSimulation(t *testing.T) { // testSyncBetweenNodes(t, 8, 1, dataChunkCount, true, 1) // // testSyncBetweenNodes(t, 8, 1, dataChunkCount, false, 1) testSyncBetweenNodes(t, 16, 1, dataChunkCount, true, 1) + // testSyncBetweenNodes(t, 32, 1, dataChunkCount, true, 1) // testSyncBetweenNodes(t, 16, 1, dataChunkCount, false, 1) } @@ -61,34 +62,49 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck ToAddr: toAddr, Services: services, } + // create context for simulation run + timeout := 30 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + // defer cancel should come before defer simulation teardown + defer cancel() + // create simulation network with the config sim, teardown, err := streamTesting.NewSimulation(conf) defer teardown() if err != nil { t.Fatal(err.Error()) } + // DEBUG: defer func() { for _, id := range sim.IDs { deliveries[id].PrintCounters(id) } - // for id, delivery := range deliveries { - // delivery.PrintCounters(id) - // } }() + // HACK: these are global variables in the test so that they are available for + // the service constructor function + // TODO: will this work with exec/docker adapter? + // localstore of nodes made available for action and check calls stores = make(map[discover.NodeID]storage.ChunkStore) - deliveries = make(map[discover.NodeID]*Delivery) + nodeIndex := make(map[discover.NodeID]int) for i, id := range sim.IDs { + nodeIndex[id] = i stores[id] = sim.Stores[i] } + deliveries = make(map[discover.NodeID]*Delivery) + // peerCount function gives the number of peer connections for a nodeID + // this is needed for the service run function to wait until + // each protocol instance runs and the streamer peers are available peerCount = func(id discover.NodeID) int { if sim.IDs[0] == id || sim.IDs[nodes-1] == id { return 1 } return 2 } - // here we distribute chunks of a random file into Stores of nodes 1 to nodes + waitPeerErrC = make(chan error) + + // here we distribute chunks of a random file into stores 1...nodes rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams()) rrdpa.Start() size := chunkCount * chunkSize @@ -100,12 +116,14 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck t.Fatal(err.Error()) } - // collect hashes in po 1 from all nodes - hashes := make([][]storage.Key, nodes) + // create DBAPI-s for all nodes dbs := make([]*storage.DBAPI, nodes) for i := 0; i < nodes; i++ { dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore)) } + + // collect hashes in po 1 bin for each node + hashes := make([][]storage.Key, nodes) totalHashes := 0 hashCounts := make([]int, nodes) for i := nodes - 1; i >= 0; i-- { @@ -120,9 +138,12 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck }) } + // errc is error channel for simulation errc := make(chan error, 1) - waitPeerErrC = make(chan error) quitC := make(chan struct{}) + defer close(quitC) + + // action is subscribe action := func(ctx context.Context) error { // need to wait till an aynchronous process registers the peers in streamer.peers // that is used by Subscribe @@ -139,24 +160,29 @@ 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, peerCount(sim.IDs[j]), errc, quitC) + for j := 0; j < nodes-1; j++ { + id := sim.IDs[j] + err := sim.CallClient(id, func(client *rpc.Client) error { + // report disconnect events to the error channel cos peers should not disconnect + err := streamTesting.WatchDisconnections(id, client, peerCount(id), errc, quitC) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + // start syncing, i.e., subscribe to upstream peers po 1 bin + sid := sim.IDs[j+1] + return client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{1}, 0, 0, Top, false) + }) if err != nil { return err } - ctx, cancel := context.WithTimeout(ctx, 1*time.Second) - defer cancel() - j++ - return client.CallContext(ctx, nil, "stream_subscribeStream", sim.IDs[j], "SYNC", []byte{1}, 0, 0, Top, false) - }, sim.IDs[0:nodes-1]...) + } + return nil } // this makes sure check is not called before the previous call finishes - checkC := make(chan struct{}) check := func(ctx context.Context, id discover.NodeID) (bool, error) { - defer func() { checkC <- struct{}{} }() - select { case err := <-errc: return false, err @@ -165,54 +191,37 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck default: } - var pass bool - var i int - log.Error("staring dbs check") - for i = nodes - 1; i >= 0; i-- { - nodeHashCount := hashCounts[i] - nodeHashFound := 0 - for j := i; j < nodes; j++ { - nodeHashes := hashes[j] - for _, key := range nodeHashes { - chunk, err := dbs[i].Get(key) - if err == storage.ErrFetching { - <-chunk.ReqC - nodeHashFound++ - } else if err == nil { - nodeHashFound++ - } else { - log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) - } + log.Error("starting dbs check", "node", id) + i := nodeIndex[id] + var total, found int + for j := i; j < nodes; j++ { + total += len(hashes[j]) + for _, key := range hashes[j] { + chunk, err := dbs[i].Get(key) + if err == storage.ErrFetching { + <-chunk.ReqC + } else if err != nil { + log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) + continue } - } - log.Error("sync check", "node", sim.IDs[i], "index", i, "bin", po, "found", nodeHashFound, "total", nodeHashCount) - pass = nodeHashFound == nodeHashCount - if !pass { - break + // needed for leveldb not to be closed? + // chunk.WaitToStore() + found++ } } - // log.Error("sync check", "bin", po, "found", found, "total", totalHashes) - // pass := found == totalHashes - if !pass { - return false, nil - } - close(quitC) - return true, nil - + log.Error("sync check", "node", id, "index", i, "bin", po, "found", found, "total", total) + return total == found, nil } conf.Step = &simulations.Step{ Action: action, - Trigger: streamTesting.PivotTrigger(500*time.Millisecond, checkC, sim.IDs[0]), + Trigger: streamTesting.Trigger(500*time.Millisecond, quitC, sim.IDs[0:nodes-1]...), Expect: &simulations.Expectation{ Nodes: sim.IDs[0:1], Check: check, }, } startedAt := time.Now() - timeout := 30 * time.Second - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() result, err := sim.Run(ctx, conf) finishedAt := time.Now() if err != nil { diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index 9585de369b..8ef750df1b 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -234,7 +234,7 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, expectedConnCou return nil } -func PivotTrigger(d time.Duration, checkC chan struct{}, ids ...discover.NodeID) chan discover.NodeID { +func Trigger(d time.Duration, quitC chan struct{}, ids ...discover.NodeID) chan discover.NodeID { trigger := make(chan discover.NodeID) go func() { ticker := time.NewTicker(d) @@ -242,28 +242,24 @@ func PivotTrigger(d time.Duration, checkC chan struct{}, ids ...discover.NodeID) // we are only testing the pivot node (net.Nodes[0]) for range ticker.C { for _, id := range ids { - trigger <- id + select { + case trigger <- id: + case <-quitC: + } } - <-checkC } }() return trigger } -func (sim *Simulation) CallClient(f func(*rpc.Client) error, ids ...discover.NodeID) error { - for _, id := range ids { - node := sim.Net.GetNode(id) - if node == nil { - return fmt.Errorf("unknown node: %s", id) - } - client, err := node.Client() - if err != nil { - return fmt.Errorf("error getting node client: %s", err) - } - err = f(client) - if err != nil { - return err - } +func (sim *Simulation) CallClient(id discover.NodeID, f func(*rpc.Client) error) error { + node := sim.Net.GetNode(id) + if node == nil { + return fmt.Errorf("unknown node: %s", id) } - return nil + client, err := node.Client() + if err != nil { + return fmt.Errorf("error getting node client: %s", err) + } + return f(client) } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index 41b0b67a24..74890b5ddf 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -781,27 +781,22 @@ func (s *DbStore) Close() { s.db.Close() } -// initialises a sync iterator from a syncToken (passed in with the handshake) +// SyncIterator(start, stop, po, f) calls f on each hash of a bin po from start to stop func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, uint64) bool) error { - // probably, the lock is not needed - // s.lock.Lock() - // defer s.lock.Unlock() - + sincekey := getDataKey(since, po) untilkey := getDataKey(until, po) - it := s.db.NewIterator() - seek := getDataKey(since, po) - it.Seek(seek) defer it.Release() + it.Seek(sincekey) for it.Next() { dbkey := it.Key() if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 { break } - key := make([]byte, 32) - copy(key, it.Value()[:32]) + val := it.Value() + copy(key, val[:32]) if !f(Key(key), binary.BigEndian.Uint64(dbkey[2:])) { break }