transport_diagnostics.go RawHistoryFolder

1package dgram
2
3import (
4	"time"
5
6	"github.com/bnema/vev/internal/ports"
7)
8
9func (t *Transport) diagnosticLoop() {
10	for {
11		select {
12		case d := <-t.diagnosticCh:
13			t.observe(d)
14		case <-t.done:
15			return
16		}
17	}
18}
19
20func (t *Transport) emitDiagnostic() {
21	if t.diagnosticCh == nil {
22		return
23	}
24	d := t.diagnosticSnapshot()
25	select {
26	case t.diagnosticCh <- d:
27	default:
28	}
29}
30
31func (t *Transport) diagnosticSnapshot() Diagnostic {
32	t.mu.Lock()
33	now := t.clock.Now()
34	lastAuthenticatedPacket := t.lastAuthenticatedPacket
35	lastCompleteRecord := t.lastCompleteRecord
36	lastACKProgress := t.lastACKProgress
37	pendingBytes := 0
38	for _, p := range t.pending {
39		if p != nil {
40			pendingBytes += len(p.frame.Payload)
41		}
42	}
43	state := t.linkState
44	pendingRecords := len(t.pending)
45	retransmits := t.retransmits
46	reassemblyInflight := t.reassemblyInflight
47	t.mu.Unlock()
48
49	return Diagnostic{
50		At:                       now,
51		State:                    state,
52		SinceAuthenticatedPacket: diagnosticAge(now, lastAuthenticatedPacket),
53		SinceCompleteRecord:      diagnosticAge(now, lastCompleteRecord),
54		SinceACKProgress:         diagnosticAge(now, lastACKProgress),
55		PendingRecords:           pendingRecords,
56		PendingBytes:             pendingBytes,
57		Retransmits:              retransmits,
58		ReassemblyInflight:       reassemblyInflight,
59	}
60}
61
62func diagnosticAge(now, then time.Time) time.Duration {
63	if then.After(now) {
64		return 0
65	}
66	return now.Sub(then)
67}
68
69func (t *Transport) beginRuntimeOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
70	if t.runtimeObserver == nil {
71		return func(bool) {}
72	}
73	if !t.beginObservedOperation() {
74		return func(bool) {}
75	}
76	correlation := ports.NewRuntimeCorrelation()
77	t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, start, bytes, true))
78	end := ports.RuntimeAdapterSendEnd
79	if start == ports.RuntimeAdapterReceiveStart {
80		end = ports.RuntimeAdapterReceiveEnd
81	}
82	return func(valid bool) {
83		defer t.finishObservedOperation()
84		t.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("dgram", correlation, end, bytes, valid))
85	}
86}
87
88func (t *Transport) beginObservedOperation() bool {
89	t.operationMu.Lock()
90	defer t.operationMu.Unlock()
91	if t.closing {
92		return false
93	}
94	if t.operationCount == 0 {
95		t.operationsDone = make(chan struct{})
96	}
97	t.operationCount++
98	return true
99}
100
101func (t *Transport) finishObservedOperation() {
102	t.operationMu.Lock()
103	defer t.operationMu.Unlock()
104	t.operationCount--
105	if t.operationCount == 0 {
106		close(t.operationsDone)
107	}
108}
109
110func (t *Transport) beginShutdown() <-chan struct{} {
111	t.operationMu.Lock()
112	defer t.operationMu.Unlock()
113	t.closing = true
114	if t.operationCount == 0 {
115		return nil
116	}
117	return t.operationsDone
118}