()
| 659 | } |
| 660 | |
| 661 | func (s *SyncClient) mainLoop() { |
| 662 | defer s.wg.Done() |
| 663 | |
| 664 | s.cleanTasks() |
| 665 | if !s.syncDone { |
| 666 | err := s.storageManager.DownloadAllMetas(s.resCtx, s.syncerParams.MetaDownloadBatchSize) |
| 667 | if err != nil { |
| 668 | s.lg.Error("Download blob metadata failed", "error", err) |
| 669 | return |
| 670 | } |
| 671 | } |
| 672 | |
| 673 | s.logTime = time.Now() |
| 674 | for { |
| 675 | // Remove all completed tasks and terminate sync if everything's done |
| 676 | s.cleanTasks() |
| 677 | if s.syncDone { |
| 678 | s.report(true) |
| 679 | s.saveSyncStatus() |
| 680 | return |
| 681 | } |
| 682 | |
| 683 | s.assignBlobRangeTasks() |
| 684 | // Assign all the Data retrieval tasks to any free peers |
| 685 | s.assignBlobHealTasks() |
| 686 | |
| 687 | s.assignFillEmptyBlobTasks() |
| 688 | |
| 689 | select { |
| 690 | case <-time.After(requestTimeoutInMillisecond): |
| 691 | |
| 692 | case <-s.update: |
| 693 | // Something happened (new peer, delivery, timeout), recheck tasks |
| 694 | case <-s.peerJoin: |
| 695 | // A new peer joined, try to schedule it new tasks |
| 696 | case <-s.resCtx.Done(): |
| 697 | s.lg.Info("Stopped P2P req-resp L2 block sync client") |
| 698 | return |
| 699 | } |
| 700 | // Report stats if something meaningful happened |
| 701 | s.report(false) |
| 702 | } |
| 703 | } |
| 704 | |
| 705 | func (s *SyncClient) notifyPeerJoin(id peer.ID) { |
| 706 | select { |
no test coverage detected