diff --git a/discovery/gossiper.go b/discovery/gossiper.go index ae6de9f71..e3bca5e1f 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -1154,34 +1154,48 @@ func (d *AuthenticatedGossiper) splitAnnouncementBatches( } // splitAndSendAnnBatch takes a batch of messages, computes the proper batch -// split size, and then sends out all items to the set of target peers. If -// isLocal is true, then we'll send them to all peers and skip any gossip -// filter checks. -func (d *AuthenticatedGossiper) splitAndSendAnnBatch(annBatch []msgWithSenders, - isLocal bool) { +// split size, and then sends out all items to the set of target peers. Locally +// generated announcements are always sent before remotely generated +// announcements. +func (d *AuthenticatedGossiper) splitAndSendAnnBatch( + annBatch msgsToBroadcast) { - splitAnnouncementBatch := d.splitAnnouncementBatches(annBatch) + // delayNextBatch is a helper closure that blocks for `SubBatchDelay` + // duration to delay the sending of next announcement batch. + delayNextBatch := func() { + select { + case <-time.After(d.cfg.SubBatchDelay): + case <-d.quit: + return + } + } + + // Fetch the local and remote announcements. + localBatches := d.splitAnnouncementBatches(annBatch.localMsgs) + remoteBatches := d.splitAnnouncementBatches(annBatch.remoteMsgs) d.wg.Add(1) go func() { defer d.wg.Done() - log.Infof("Broadcasting %v new announcements in %d sub "+ - "batches (local=%v)", len(annBatch), - len(splitAnnouncementBatch), isLocal) + log.Debugf("Broadcasting %v new local announcements in %d "+ + "sub batches", len(annBatch.localMsgs), + len(localBatches)) - for _, announcementBatch := range splitAnnouncementBatch { - if isLocal { - d.sendLocalBatch(announcementBatch) - } else { - d.sendRemoteBatch(announcementBatch) - } + // Send out the local announcements first. + for _, annBatch := range localBatches { + d.sendLocalBatch(annBatch) + delayNextBatch() + } - select { - case <-time.After(d.cfg.SubBatchDelay): - case <-d.quit: - return - } + log.Debugf("Broadcasting %v new remote announcements in %d "+ + "sub batches", len(annBatch.remoteMsgs), + len(remoteBatches)) + + // Now send the remote announcements. + for _, annBatch := range remoteBatches { + d.sendRemoteBatch(annBatch) + delayNextBatch() } }() } @@ -1354,12 +1368,7 @@ func (d *AuthenticatedGossiper) networkHandler() { // announcements, we'll blast them out w/o regard for // our peer's policies so we ensure they propagate // properly. - d.splitAndSendAnnBatch( - announcementBatch.localMsgs, true, - ) - d.splitAndSendAnnBatch( - announcementBatch.remoteMsgs, false, - ) + d.splitAndSendAnnBatch(announcementBatch) // The retransmission timer has ticked which indicates that we // should check if we need to prune or re-broadcast any of our diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index c8ed6a14d..322cd9564 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -784,7 +784,7 @@ func createTestCtx(t *testing.T, startHeight uint32) (*testCtx, error) { HistoricalSyncTicker: ticker.NewForce(DefaultHistoricalSyncInterval), NumActiveSyncers: 3, AnnSigner: &mock.SingleSigner{Privkey: selfKeyPriv}, - SubBatchDelay: time.Second * 5, + SubBatchDelay: 1 * time.Millisecond, MinimumBatchSize: 10, MaxChannelUpdateBurst: DefaultMaxChannelUpdateBurst, ChannelUpdateInterval: DefaultChannelUpdateInterval, @@ -3371,7 +3371,9 @@ out: case <-sentMsgs: case err := <-notifyErr: t.Fatal(err) - default: + + // Give it 5 seconds to drain out. + case <-time.After(5 * time.Second): break out } }