transport_health.go RawHistoryFolder

1package dgram
2
3import (
4	"time"
5
6	"github.com/bnema/vev/internal/ports"
7)
8
9// healthLoop owns link-health timing so bulk retransmission cannot delay state
10// transitions or socket hopping.
11func (t *Transport) healthLoop() {
12	health := t.clock.NewTimer(t.resendAfter)
13	defer health.Stop()
14	for {
15		select {
16		case <-health.C():
17			t.checkSilence()
18			health.Reset(t.resendAfter)
19			t.notifyTimerHook(func() func() { return t.afterHealthTimer })
20		case <-t.done:
21			return
22		}
23	}
24}
25
26func (t *Transport) heartbeatLoop() {
27	heartbeat := t.clock.NewTimer(t.heartbeat)
28	defer heartbeat.Stop()
29	for {
30		select {
31		case <-heartbeat.C():
32			t.queueControl(recProbe, 0)
33			heartbeat.Reset(t.heartbeat)
34			t.notifyTimerHook(func() func() { return t.afterHeartbeatTimer })
35		case <-t.done:
36			return
37		}
38	}
39}
40
41func (t *Transport) notifyTimerHook(get func() func()) {
42	t.mu.Lock()
43	hook := get()
44	t.mu.Unlock()
45	if hook != nil {
46		hook()
47	}
48}
49
50func (t *Transport) hopPacketConnOnce(generation uint64) {
51	t.mu.Lock()
52	if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
53		t.mu.Unlock()
54		return
55	}
56	old := t.pc
57	rebind := t.rebind
58	t.hoppedOffline = true
59	t.hopGeneration = generation
60	t.mu.Unlock()
61	pc, err := rebind(old)
62
63	// Lock order is writeMu -> controlConnMu -> mu. Data writes take only
64	// writeMu, while control writes hold controlConnMu's read lock without
65	// writeMu; therefore a hop waits for active control writes but does not make
66	// them wait for a blocked data write.
67	t.writeMu.Lock()
68	defer t.writeMu.Unlock()
69	t.controlConnMu.Lock()
70	defer t.controlConnMu.Unlock()
71	t.mu.Lock()
72	if err != nil {
73		t.rollbackHopLocked(generation)
74		t.mu.Unlock()
75		return
76	}
77	if t.closed || t.pc != old || t.health.generation != generation || t.linkState != ports.LinkStateProbing {
78		t.rollbackHopLocked(generation)
79		t.mu.Unlock()
80		_ = pc.Close()
81		return
82	}
83	t.pc = pc
84	t.writeDeadlines = newWriteDeadlineState(pc)
85	t.mu.Unlock()
86	go t.readLoop(pc)
87	_ = old.Close()
88}
89
90func (t *Transport) rollbackHopLocked(generation uint64) {
91	if t.hoppedOffline && t.hopGeneration == generation {
92		t.hoppedOffline = false
93		t.hopGeneration = 0
94	}
95}
96
97func (t *Transport) checkSilence() {
98	runHook := true
99	for {
100		t.mu.Lock()
101		if t.closed {
102			t.mu.Unlock()
103			return
104		}
105		now := t.clock.Now()
106		generation := t.health.generation
107		state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
108		afterDecision := t.afterHealthDecision
109		t.mu.Unlock()
110
111		if runHook {
112			runHook = false
113			if afterDecision != nil {
114				afterDecision()
115			}
116		}
117
118		t.mu.Lock()
119		if t.closed {
120			t.mu.Unlock()
121			return
122		}
123		if t.health.generation != generation {
124			t.mu.Unlock()
125			continue
126		}
127		changed, stateGeneration := t.setLinkStateLocked(state)
128		afterStateCommit := t.afterLinkStateCommit
129		closed := false
130		if dead {
131			closed = t.closeWithErrorLocked(ErrLinkDead)
132		}
133		t.mu.Unlock()
134
135		if changed {
136			if afterStateCommit != nil {
137				afterStateCommit(state)
138			}
139			var linkErr error
140			if dead {
141				linkErr = ErrLinkDead
142			}
143			t.publishLinkState(state, now, linkErr, stateGeneration)
144		}
145		if closed {
146			t.broadcastClosed()
147			return
148		}
149		if hop {
150			t.hopPacketConnOnce(generation)
151		}
152		return
153	}
154}
155
156func (t *Transport) setLinkState(state ports.LinkState, err error) {
157	now := t.clock.Now()
158	t.mu.Lock()
159	changed, stateGeneration := t.setLinkStateLocked(state)
160	afterStateCommit := t.afterLinkStateCommit
161	t.mu.Unlock()
162	if changed {
163		if afterStateCommit != nil {
164			afterStateCommit(state)
165		}
166		t.publishLinkState(state, now, err, stateGeneration)
167	}
168}
169
170func (t *Transport) setLinkStateLocked(state ports.LinkState) (bool, uint64) {
171	if t.linkState == state {
172		return false, t.linkStateGeneration
173	}
174	t.linkState = state
175	t.linkStateGeneration++
176	if state == ports.LinkStateConnected {
177		t.hoppedOffline = false
178		t.hopGeneration = 0
179		for _, p := range t.pending {
180			if p != nil {
181				p.attempts = 0
182			}
183		}
184	}
185	t.notifySendWaitersLocked()
186	return true, t.linkStateGeneration
187}
188
189func (t *Transport) publishLinkState(state ports.LinkState, now time.Time, err error, generation uint64) {
190	t.linkEventMu.Lock()
191	defer t.linkEventMu.Unlock()
192	t.mu.Lock()
193	if t.linkState != state || t.linkStateGeneration != generation {
194		t.mu.Unlock()
195		return
196	}
197	event := ports.LinkEvent{State: state, At: now, Err: err}
198	select {
199	case t.linkEvents <- event:
200	default:
201	}
202	t.mu.Unlock()
203	t.emitDiagnostic()
204}