mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-25 14:16:44 +00:00
fix close sequence
It is not guaranteed that Next will be called until exhaustion after Close was called. Hence, we need to empty override the passed channel. Signed-off-by: Csaba Kiraly <csaba.kiraly@gmail.com>
This commit is contained in:
parent
c1af08f649
commit
e7ea81d945
1 changed files with 7 additions and 2 deletions
|
|
@ -155,7 +155,7 @@ func (f *filterIter) Next() bool {
|
||||||
// AsyncFilter wraps an iterator such that Next only returns nodes for which
|
// AsyncFilter wraps an iterator such that Next only returns nodes for which
|
||||||
// the 'check' function returns a (possibly modified) node.
|
// the 'check' function returns a (possibly modified) node.
|
||||||
func AsyncFilter(it Iterator, check func(*Node) (*Node, error), workers int) Iterator {
|
func AsyncFilter(it Iterator, check func(*Node) (*Node, error), workers int) Iterator {
|
||||||
f := &AsyncFilterIter{it, check, make(chan struct{}, workers), make(chan *Node), nil}
|
f := &AsyncFilterIter{it, check, make(chan struct{}, workers), make(chan *Node), make(chan struct{}), nil}
|
||||||
|
|
||||||
// create slots
|
// create slots
|
||||||
for range cap(f.slots) {
|
for range cap(f.slots) {
|
||||||
|
|
@ -172,7 +172,10 @@ func AsyncFilter(it Iterator, check func(*Node) (*Node, error), workers int) Ite
|
||||||
// check the node async, in a separate goroutine
|
// check the node async, in a separate goroutine
|
||||||
go func() {
|
go func() {
|
||||||
if nn, err := f.check(n); err == nil {
|
if nn, err := f.check(n); err == nil {
|
||||||
f.passed <- nn
|
select {
|
||||||
|
case f.passed <- nn:
|
||||||
|
case <-f.closed: // bale out if downstream is already closed and not calling Next
|
||||||
|
}
|
||||||
}
|
}
|
||||||
f.slots <- struct{}{}
|
f.slots <- struct{}{}
|
||||||
}()
|
}()
|
||||||
|
|
@ -197,6 +200,7 @@ type AsyncFilterIter struct {
|
||||||
check func(*Node) (*Node, error)
|
check func(*Node) (*Node, error)
|
||||||
slots chan struct{}
|
slots chan struct{}
|
||||||
passed chan *Node
|
passed chan *Node
|
||||||
|
closed chan struct{}
|
||||||
buffer *Node
|
buffer *Node
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -211,6 +215,7 @@ func (f *AsyncFilterIter) Node() *Node {
|
||||||
|
|
||||||
func (f *AsyncFilterIter) Close() {
|
func (f *AsyncFilterIter) Close() {
|
||||||
f.it.Close()
|
f.it.Close()
|
||||||
|
close(f.closed) // override the passed channel
|
||||||
for range cap(f.slots) {
|
for range cap(f.slots) {
|
||||||
<-f.slots
|
<-f.slots
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue