forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

core/bloombits/matcher | Remove the kill channel to quit directly for matcher session

Browse source
This commit is contained in:
ucwong 2020-04-08 09:53:12 +00:00
parent 3ec95e7751
commit 3ed5d5500f
1 changed files with 0 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
core/bloombits/matcher.go
View file

@ -155,7 +155,6 @@ func (m *Matcher) Start(ctx context.Context, begin, end uint64, results chan uin
session := &MatcherSession{ session := &MatcherSession{
matcher: m, matcher: m,
quit: make(chan struct{}), quit: make(chan struct{}),
kill: make(chan struct{}),
ctx: ctx, ctx: ctx,
} }
for _, scheduler := range m.schedulers { for _, scheduler := range m.schedulers {
@ -415,10 +414,6 @@ func (m *Matcher) distributor(dist chan *request, session *MatcherSession) {
} }
shutdown = nil shutdown = nil
case <-session.kill:
// Pending requests not honoured in time, hard terminate
return
case req := <-dist: case req := <-dist:
// New retrieval request arrived to be distributed to some fetcher process // New retrieval request arrived to be distributed to some fetcher process
queue := requests[req.bit] queue := requests[req.bit]
@ -514,7 +509,6 @@ type MatcherSession struct {
closer sync.Once // Sync object to ensure we only ever close once closer sync.Once // Sync object to ensure we only ever close once
quit chan struct{} // Quit channel to request pipeline termination quit chan struct{} // Quit channel to request pipeline termination
kill chan struct{} // Term channel to signal non-graceful forced shutdown
ctx context.Context // Context used by the light client to abort filtering ctx context.Context // Context used by the light client to abort filtering
err atomic.Value // Global error to track retrieval failures deep in the chain err atomic.Value // Global error to track retrieval failures deep in the chain
@ -529,8 +523,6 @@ func (s *MatcherSession) Close() {
s.closer.Do(func() { s.closer.Do(func() {
// Signal termination and wait for all goroutines to tear down // Signal termination and wait for all goroutines to tear down
close(s.quit) close(s.quit)
timeout := time.AfterFunc(time.Second, func() { close(s.kill) })
defer timeout.Stop()
s.pend.Wait() s.pend.Wait()
}) })
} }
@ -594,8 +586,6 @@ func (s *MatcherSession) AllocateSections(bit uint, count int) []uint64 {
// bit index to be injected into the processing pipeline. // bit index to be injected into the processing pipeline.
func (s *MatcherSession) DeliverSections(bit uint, sections []uint64, bitsets [][]byte) { func (s *MatcherSession) DeliverSections(bit uint, sections []uint64, bitsets [][]byte) {
select { select {
case <-s.kill:
return
case s.matcher.deliveries <- &Retrieval{Bit: bit, Sections: sections, Bitsets: bitsets}: case s.matcher.deliveries <- &Retrieval{Bit: bit, Sections: sections, Bitsets: bitsets}:
} }
} }