aboutsummaryrefslogtreecommitdiffstats
path: root/eth
diff options
context:
space:
mode:
Diffstat (limited to 'eth')
-rw-r--r--eth/db_upgrade.go13
-rw-r--r--eth/filters/api.go63
-rw-r--r--eth/filters/filter_system.go19
-rw-r--r--eth/handler.go7
4 files changed, 74 insertions, 28 deletions
diff --git a/eth/db_upgrade.go b/eth/db_upgrade.go
index 12de60fe7..172bb0954 100644
--- a/eth/db_upgrade.go
+++ b/eth/db_upgrade.go
@@ -93,6 +93,9 @@ func upgradeSequentialKeys(db ethdb.Database) (stopFn func()) {
func upgradeSequentialCanonicalNumbers(db ethdb.Database, stopFn func() bool) (error, bool) {
prefix := []byte("block-num-")
it := db.(*ethdb.LDBDatabase).NewIterator()
+ defer func() {
+ it.Release()
+ }()
it.Seek(prefix)
cnt := 0
for bytes.HasPrefix(it.Key(), prefix) {
@@ -100,6 +103,9 @@ func upgradeSequentialCanonicalNumbers(db ethdb.Database, stopFn func() bool) (e
if len(keyPtr) < 20 {
cnt++
if cnt%100000 == 0 {
+ it.Release()
+ it = db.(*ethdb.LDBDatabase).NewIterator()
+ it.Seek(keyPtr)
glog.V(logger.Info).Infof("converting %d canonical numbers...", cnt)
}
number := big.NewInt(0).SetBytes(keyPtr[10:]).Uint64()
@@ -130,6 +136,9 @@ func upgradeSequentialCanonicalNumbers(db ethdb.Database, stopFn func() bool) (e
func upgradeSequentialBlocks(db ethdb.Database, stopFn func() bool) (error, bool) {
prefix := []byte("block-")
it := db.(*ethdb.LDBDatabase).NewIterator()
+ defer func() {
+ it.Release()
+ }()
it.Seek(prefix)
cnt := 0
for bytes.HasPrefix(it.Key(), prefix) {
@@ -137,6 +146,9 @@ func upgradeSequentialBlocks(db ethdb.Database, stopFn func() bool) (error, bool
if len(keyPtr) >= 38 {
cnt++
if cnt%10000 == 0 {
+ it.Release()
+ it = db.(*ethdb.LDBDatabase).NewIterator()
+ it.Seek(keyPtr)
glog.V(logger.Info).Infof("converting %d blocks...", cnt)
}
// convert header, body, td and block receipts
@@ -175,6 +187,7 @@ func upgradeSequentialBlocks(db ethdb.Database, stopFn func() bool) (error, bool
func upgradeSequentialOrphanedReceipts(db ethdb.Database, stopFn func() bool) (error, bool) {
prefix := []byte("receipts-block-")
it := db.(*ethdb.LDBDatabase).NewIterator()
+ defer it.Release()
it.Seek(prefix)
cnt := 0
for bytes.HasPrefix(it.Key(), prefix) {
diff --git a/eth/filters/api.go b/eth/filters/api.go
index 393019f8b..65c5b9380 100644
--- a/eth/filters/api.go
+++ b/eth/filters/api.go
@@ -68,8 +68,6 @@ type PublicFilterAPI struct {
transactionMu sync.RWMutex
transactionQueue map[int]*hashQueue
-
- transactMu sync.Mutex
}
// NewPublicFilterAPI returns a new PublicFilterAPI instance.
@@ -100,6 +98,7 @@ done:
for {
select {
case <-timer.C:
+ s.filterManager.Lock() // lock order like filterLoop()
s.logMu.Lock()
for id, filter := range s.logQueue {
if time.Since(filter.timeout) > filterTickerTime {
@@ -126,6 +125,7 @@ done:
}
}
s.transactionMu.Unlock()
+ s.filterManager.Unlock()
case <-s.quit:
break done
}
@@ -135,19 +135,24 @@ done:
// NewBlockFilter create a new filter that returns blocks that are included into the canonical chain.
func (s *PublicFilterAPI) NewBlockFilter() (string, error) {
+ // protect filterManager.Add() and setting of filter fields
+ s.filterManager.Lock()
+ defer s.filterManager.Unlock()
+
externalId, err := newFilterId()
if err != nil {
return "", err
}
- s.blockMu.Lock()
filter := New(s.chainDb)
id, err := s.filterManager.Add(filter, ChainFilter)
if err != nil {
return "", err
}
+ s.blockMu.Lock()
s.blockQueue[id] = &hashQueue{timeout: time.Now()}
+ s.blockMu.Unlock()
filter.BlockCallback = func(block *types.Block, logs vm.Logs) {
s.blockMu.Lock()
@@ -158,8 +163,6 @@ func (s *PublicFilterAPI) NewBlockFilter() (string, error) {
}
}
- defer s.blockMu.Unlock()
-
s.filterMapMu.Lock()
s.filterMapping[externalId] = id
s.filterMapMu.Unlock()
@@ -169,21 +172,24 @@ func (s *PublicFilterAPI) NewBlockFilter() (string, error) {
// NewPendingTransactionFilter creates a filter that returns new pending transactions.
func (s *PublicFilterAPI) NewPendingTransactionFilter() (string, error) {
+ // protect filterManager.Add() and setting of filter fields
+ s.filterManager.Lock()
+ defer s.filterManager.Unlock()
+
externalId, err := newFilterId()
if err != nil {
return "", err
}
- s.transactionMu.Lock()
- defer s.transactionMu.Unlock()
-
filter := New(s.chainDb)
id, err := s.filterManager.Add(filter, PendingTxFilter)
if err != nil {
return "", err
}
+ s.transactionMu.Lock()
s.transactionQueue[id] = &hashQueue{timeout: time.Now()}
+ s.transactionMu.Unlock()
filter.TransactionCallback = func(tx *types.Transaction) {
s.transactionMu.Lock()
@@ -203,8 +209,9 @@ func (s *PublicFilterAPI) NewPendingTransactionFilter() (string, error) {
// newLogFilter creates a new log filter.
func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []common.Address, topics [][]common.Hash, callback func(log *vm.Log, removed bool)) (int, error) {
- s.logMu.Lock()
- defer s.logMu.Unlock()
+ // protect filterManager.Add() and setting of filter fields
+ s.filterManager.Lock()
+ defer s.filterManager.Unlock()
filter := New(s.chainDb)
id, err := s.filterManager.Add(filter, LogFilter)
@@ -212,7 +219,9 @@ func (s *PublicFilterAPI) newLogFilter(earliest, latest int64, addresses []commo
return 0, err
}
+ s.logMu.Lock()
s.logQueue[id] = &logQueue{timeout: time.Now()}
+ s.logMu.Unlock()
filter.SetBeginBlock(earliest)
filter.SetEndBlock(latest)
@@ -443,35 +452,43 @@ func (s *PublicFilterAPI) GetLogs(args NewFilterArgs) []vmlog {
// UninstallFilter removes the filter with the given filter id.
func (s *PublicFilterAPI) UninstallFilter(filterId string) bool {
- s.filterMapMu.Lock()
- defer s.filterMapMu.Unlock()
+ s.filterManager.Lock()
+ defer s.filterManager.Unlock()
+ s.filterMapMu.Lock()
id, ok := s.filterMapping[filterId]
if !ok {
+ s.filterMapMu.Unlock()
return false
}
-
- defer s.filterManager.Remove(id)
delete(s.filterMapping, filterId)
+ s.filterMapMu.Unlock()
+ s.filterManager.Remove(id)
+
+ s.logMu.Lock()
if _, ok := s.logQueue[id]; ok {
- s.logMu.Lock()
- defer s.logMu.Unlock()
delete(s.logQueue, id)
+ s.logMu.Unlock()
return true
}
+ s.logMu.Unlock()
+
+ s.blockMu.Lock()
if _, ok := s.blockQueue[id]; ok {
- s.blockMu.Lock()
- defer s.blockMu.Unlock()
delete(s.blockQueue, id)
+ s.blockMu.Unlock()
return true
}
+ s.blockMu.Unlock()
+
+ s.transactionMu.Lock()
if _, ok := s.transactionQueue[id]; ok {
- s.transactionMu.Lock()
- defer s.transactionMu.Unlock()
delete(s.transactionQueue, id)
+ s.transactionMu.Unlock()
return true
}
+ s.transactionMu.Unlock()
return false
}
@@ -525,7 +542,9 @@ func (s *PublicFilterAPI) logFilterChanged(id int) []vmlog {
// GetFilterLogs returns the logs for the filter with the given id.
func (s *PublicFilterAPI) GetFilterLogs(filterId string) []vmlog {
+ s.filterMapMu.RLock()
id, ok := s.filterMapping[filterId]
+ s.filterMapMu.RUnlock()
if !ok {
return toRPCLogs(nil, false)
}
@@ -540,9 +559,9 @@ func (s *PublicFilterAPI) GetFilterLogs(filterId string) []vmlog {
// GetFilterChanges returns the logs for the filter with the given id since last time is was called.
// This can be used for polling.
func (s *PublicFilterAPI) GetFilterChanges(filterId string) interface{} {
- s.filterMapMu.Lock()
+ s.filterMapMu.RLock()
id, ok := s.filterMapping[filterId]
- s.filterMapMu.Unlock()
+ s.filterMapMu.RUnlock()
if !ok { // filter not found
return []interface{}{}
diff --git a/eth/filters/filter_system.go b/eth/filters/filter_system.go
index 4343dfa21..256464213 100644
--- a/eth/filters/filter_system.go
+++ b/eth/filters/filter_system.go
@@ -82,11 +82,20 @@ func (fs *FilterSystem) Stop() {
fs.sub.Unsubscribe()
}
-// Add adds a filter to the filter manager
-func (fs *FilterSystem) Add(filter *Filter, filterType FilterType) (int, error) {
+// Acquire filter system maps lock, required to force lock acquisition
+// sequence with filterMu acquired first to avoid deadlocks by callbacks
+func (fs *FilterSystem) Lock() {
fs.filterMu.Lock()
- defer fs.filterMu.Unlock()
+}
+
+// Release filter system maps lock
+func (fs *FilterSystem) Unlock() {
+ fs.filterMu.Unlock()
+}
+// Add adds a filter to the filter manager
+// Expects filterMu to be locked.
+func (fs *FilterSystem) Add(filter *Filter, filterType FilterType) (int, error) {
id := fs.filterId
filter.created = time.Now()
@@ -110,10 +119,8 @@ func (fs *FilterSystem) Add(filter *Filter, filterType FilterType) (int, error)
}
// Remove removes a filter by filter id
+// Expects filterMu to be locked.
func (fs *FilterSystem) Remove(id int) {
- fs.filterMu.Lock()
- defer fs.filterMu.Unlock()
-
delete(fs.chainFilters, id)
delete(fs.pendingTxFilters, id)
delete(fs.logFilters, id)
diff --git a/eth/handler.go b/eth/handler.go
index 5bed63cc0..9ad430976 100644
--- a/eth/handler.go
+++ b/eth/handler.go
@@ -294,6 +294,13 @@ func (pm *ProtocolManager) handle(p *peer) error {
glog.V(logger.Warn).Infof("%v: timed out DAO fork-check, dropping", p)
pm.removePeer(p.id)
})
+ // Make sure it's cleaned up if the peer dies off
+ defer func() {
+ if p.forkDrop != nil {
+ p.forkDrop.Stop()
+ p.forkDrop = nil
+ }
+ }()
}
// main loop. handle incoming messages.
for {
generated by cgit v1.2.3 (git 2.25.1) at 2024-08-25 23:07:16 +0800