1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
|
// Copyright (c) Tailscale Inc & contributors
// SPDX-License-Identifier: BSD-3-Clause
package ipnlocal
import (
"context"
"time"
"tailscale.com/ipn"
"tailscale.com/tailcfg"
"tailscale.com/tstime"
)
type rateLimitingBusSender struct {
fn func(*ipn.Notify) (keepGoing bool)
lastFlush time.Time // last call to fn, or zero value if none
interval time.Duration // 0 to flush immediately; non-zero to rate limit sends
clock tstime.DefaultClock // non-nil for testing
didSendTestHook func() // non-nil for testing
// pending, if non-nil, is the pending notification that we
// haven't sent yet. We own this memory to mutate.
pending *ipn.Notify
// flushTimer is non-nil if the timer is armed.
flushTimer tstime.TimerController // effectively a *time.Timer
flushTimerC <-chan time.Time // ... said ~Timer's C chan
}
func (s *rateLimitingBusSender) close() {
if s.flushTimer != nil {
s.flushTimer.Stop()
}
}
func (s *rateLimitingBusSender) flushChan() <-chan time.Time {
return s.flushTimerC
}
func (s *rateLimitingBusSender) flush() (keepGoing bool) {
if n := s.pending; n != nil {
s.pending = nil
return s.flushNotify(n)
}
return true
}
func (s *rateLimitingBusSender) flushNotify(n *ipn.Notify) (keepGoing bool) {
s.lastFlush = s.clock.Now()
return s.fn(n)
}
// send conditionally sends n to the underlying fn, possibly rate
// limiting it, depending on whether s.interval is set, and whether
// n is a notable notification that the client (typically a GUI) would
// want to act on (render) immediately.
//
// It returns whether the caller should keep looping.
//
// The passed-in memory 'n' is owned by the caller and should
// not be mutated.
func (s *rateLimitingBusSender) send(n *ipn.Notify) (keepGoing bool) {
if s.interval <= 0 {
// No rate limiting case.
return s.fn(n)
}
if isNotableNotify(n) {
// Notable notifications are always sent immediately.
// But first send any boring one that was pending.
// TODO(bradfitz): there might be a boring one pending
// with a NetMap or Engine field that is redundant
// with the new one (n) with NetMap or Engine populated.
// We should clear the pending one's NetMap/Engine in
// that case. Or really, merge the two, but mergeBoringNotifies
// only handles the case of both sides being boring.
// So for now, flush both.
if !s.flush() {
return false
}
return s.flushNotify(n)
}
s.pending = mergeBoringNotifies(s.pending, n)
d := s.clock.Now().Sub(s.lastFlush)
if d > s.interval {
return s.flush()
}
nextFlushIn := s.interval - d
if s.flushTimer == nil {
s.flushTimer, s.flushTimerC = s.clock.NewTimer(nextFlushIn)
} else {
s.flushTimer.Reset(nextFlushIn)
}
return true
}
func (s *rateLimitingBusSender) Run(ctx context.Context, ch <-chan *ipn.Notify) {
for {
select {
case <-ctx.Done():
return
case n, ok := <-ch:
if !ok {
return
}
if !s.send(n) {
return
}
if f := s.didSendTestHook; f != nil {
f()
}
case <-s.flushChan():
if !s.flush() {
return
}
}
}
}
// mergeBoringNotify merges new notify src into possibly-nil dst,
// either mutating dst or allocating a new one if dst is nil,
// returning the merged result.
//
// dst and src must both be "boring" (i.e. not notable per isNotifiableNotify).
func mergeBoringNotifies(dst, src *ipn.Notify) *ipn.Notify {
if dst == nil {
dst = &ipn.Notify{Version: src.Version}
}
if src.NetMap != nil {
dst.NetMap = src.NetMap
dst.PeerChanges = nil // full netmap supersedes any accumulated deltas
} else if src.PeerChanges != nil {
dst.PeerChanges = mergePeerChanges(dst.PeerChanges, src.PeerChanges)
}
if src.Engine != nil {
dst.Engine = src.Engine
}
return dst
}
// mergePeerChanges merges new peer changes from src into dst, either
// mutating dst or allocating a new slice if dst is nil, returning the merged result.
// Values in src override those in dst for the same NodeID.
func mergePeerChanges(dst, src []*tailcfg.PeerChange) []*tailcfg.PeerChange {
idxByNode := make(map[tailcfg.NodeID]int, len(dst))
for i, d := range dst {
idxByNode[d.NodeID] = i
}
for _, nd := range src {
if oi, ok := idxByNode[nd.NodeID]; ok {
dst[oi] = mergePeerChangeForIpnBus(dst[oi], nd)
continue
}
idxByNode[nd.NodeID] = len(dst)
dst = append(dst, nd)
}
return dst
}
// mergePeerChangeForIpnBus merges new with old, returning the result.
// Fields set in new override those in old; fields only set in old are preserved.
func mergePeerChangeForIpnBus(old, new *tailcfg.PeerChange) *tailcfg.PeerChange {
merged := *old
// This is a subset of PeerChange that reflects only the fields that can
// be changed via a NodeMutation. If future fields can be updated via
// NodeMutations from map responses (and they are relevant to the ipn bus), then
// they should be added here and merged in the same way.
if new.DERPRegion != 0 {
// netmap.NodeMutationDerpHome
merged.DERPRegion = new.DERPRegion
}
if new.Online != nil {
// netmap.NodeMutationOnline
merged.Online = new.Online
}
if new.LastSeen != nil {
// netmap.NodeMutationLastSeen
merged.LastSeen = new.LastSeen
}
if new.Endpoints != nil {
// netmap.NodeMutationEndpoints
merged.Endpoints = new.Endpoints
}
return &merged
}
// isNotableNotify reports whether n is a "notable" notification that
// should be sent on the IPN bus immediately (e.g. to GUIs) without
// rate limiting it for a few seconds.
//
// It effectively reports whether n contains any field set that's
// not NetMap or Engine.
func isNotableNotify(n *ipn.Notify) bool {
if n == nil {
return false
}
return n.State != nil ||
n.SessionID != "" ||
n.BrowseToURL != nil ||
n.LocalTCPPort != nil ||
n.ClientVersion != nil ||
n.Prefs != nil ||
n.ErrMessage != nil ||
n.LoginFinished != nil ||
!n.DriveShares.IsNil() ||
n.Health != nil ||
len(n.IncomingFiles) > 0 ||
len(n.OutgoingFiles) > 0 ||
n.FilesWaiting != nil ||
n.SuggestedExitNode != nil
}
|
