-
-
Notifications
You must be signed in to change notification settings - Fork 273
/
msg.go
106 lines (88 loc) Β· 2.28 KB
/
msg.go
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
package terminal
import (
"fmt"
"runtime"
"github.com/safing/portbase/container"
"github.com/safing/portmaster/spn/unit"
)
// Msg is a message within the SPN network stack.
// It includes metadata and unit scheduling.
type Msg struct {
FlowID uint32
Type MsgType
Data *container.Container
// Unit scheduling.
// Note: With just 100B per packet, a uint64 (the Unit ID) is enough for
// over 1800 Exabyte. No need for overflow support.
Unit *unit.Unit
}
// NewMsg returns a new msg.
// The FlowID is unset.
// The Type is Data.
func NewMsg(data []byte) *Msg {
msg := &Msg{
Type: MsgTypeData,
Data: container.New(data),
Unit: scheduler.NewUnit(),
}
// Debug unit leaks.
msg.debugWithCaller(2)
return msg
}
// NewEmptyMsg returns a new empty msg with an initialized Unit.
// The FlowID is unset.
// The Type is Data.
// The Data is unset.
func NewEmptyMsg() *Msg {
msg := &Msg{
Type: MsgTypeData,
Unit: scheduler.NewUnit(),
}
// Debug unit leaks.
msg.debugWithCaller(2)
return msg
}
// Pack prepends the message header (Length and ID+Type) to the data.
func (msg *Msg) Pack() {
MakeMsg(msg.Data, msg.FlowID, msg.Type)
}
// Consume adds another Message to itself.
// The given Msg is packed before adding it to the data.
// The data is moved - not copied!
// High priority mark is inherited.
func (msg *Msg) Consume(other *Msg) {
// Pack message to be added.
other.Pack()
// Move data.
msg.Data.AppendContainer(other.Data)
// Inherit high priority.
if other.Unit.IsHighPriority() {
msg.Unit.MakeHighPriority()
}
// Finish other unit.
other.Finish()
}
// Finish signals the unit scheduler that this unit has finished processing.
// Will no-op if called on a nil Msg.
func (msg *Msg) Finish() {
// Proxying is necessary, as a nil msg still panics.
if msg == nil {
return
}
msg.Unit.Finish()
}
// Debug registers the unit for debug output with the given source.
// Additional calls on the same unit update the unit source.
// StartDebugLog() must be called before calling DebugUnit().
func (msg *Msg) Debug() {
msg.debugWithCaller(2)
}
func (msg *Msg) debugWithCaller(skip int) { //nolint:unparam
if !debugUnitScheduling || msg == nil {
return
}
_, file, line, ok := runtime.Caller(skip)
if ok {
scheduler.DebugUnit(msg.Unit, fmt.Sprintf("%s:%d", file, line))
}
}