| 552 | } |
| 553 | |
| 554 | func runPeerListUpdater() (err error) { |
| 555 | var privKey *asymmetric.PrivateKey |
| 556 | if privKey, err = kms.GetLocalPrivateKey(); err != nil { |
| 557 | return |
| 558 | } |
| 559 | |
| 560 | if !atomic.CompareAndSwapUint32(&peersUpdaterRunning, 0, 1) { |
| 561 | return |
| 562 | } |
| 563 | |
| 564 | go func() { |
| 565 | for { |
| 566 | if atomic.LoadUint32(&peersUpdaterRunning) == 0 { |
| 567 | return |
| 568 | } |
| 569 | |
| 570 | var wg sync.WaitGroup |
| 571 | |
| 572 | peerList.Range(func(rawDBID, _ interface{}) bool { |
| 573 | dbID := rawDBID.(proto.DatabaseID) |
| 574 | |
| 575 | wg.Add(1) |
| 576 | go func(dbID proto.DatabaseID) { |
| 577 | defer wg.Done() |
| 578 | var err error |
| 579 | |
| 580 | if _, err = getPeers(dbID, privKey); err != nil { |
| 581 | log.WithField("db", dbID). |
| 582 | WithError(err). |
| 583 | Debug("update peers failed") |
| 584 | |
| 585 | // TODO(xq262144), better rpc remote error judgement |
| 586 | if strings.Contains(err.Error(), bp.ErrNoSuchDatabase.Error()) { |
| 587 | log.WithField("db", dbID). |
| 588 | Warning("database no longer exists, stopping peers update") |
| 589 | peerList.Delete(dbID) |
| 590 | } |
| 591 | } |
| 592 | }(dbID) |
| 593 | |
| 594 | return true |
| 595 | }) |
| 596 | |
| 597 | wg.Wait() |
| 598 | |
| 599 | time.Sleep(PeersUpdateInterval) |
| 600 | } |
| 601 | }() |
| 602 | |
| 603 | return |
| 604 | } |
| 605 | |
| 606 | func stopPeersUpdater() { |
| 607 | atomic.StoreUint32(&peersUpdaterRunning, 0) |