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
|
// Package uevent provides userspace client for consuming events from a
// NETLINK_KOBJECT_UEVENT socket, as well as helpers for supplementing
// events received from the kernel.
package uevent
import (
"context"
"errors"
"strconv"
"sync/atomic"
"syscall"
"hakurei.app/internal/netlink"
)
type (
// Recoverable is satisfied by errors that are safe to recover from.
Recoverable interface{ recoverable() }
// Nontrivial is satisfied by errors preferring a JSON encoding.
Nontrivial interface{ nontrivial() }
// NeedsColdboot is satisfied by errors indicating divergence of local state
// from the kernel, usually from lost uevent data.
NeedsColdboot interface {
Recoverable
coldboot()
}
)
const (
exclConsume = iota
_exclLen
)
// Conn represents a NETLINK_KOBJECT_UEVENT socket.
type Conn struct {
conn *netlink.Conn
// Whether currently between a call to enterExcl and exitExcl.
excl [_exclLen]atomic.Bool
}
// enterExcl must be called entering a critical section that interacts with conn.
func (c *Conn) enterExcl(k int) error {
if !c.excl[k].CompareAndSwap(false, true) {
return syscall.EAGAIN
}
return nil
}
// exitExcl must be called exiting a critical section that interacts with conn.
func (c *Conn) exitExcl(k int) { c.excl[k].Store(false) }
// Close closes the underlying socket.
func (c *Conn) Close() error { return c.conn.Close() }
// Dial returns the address of a newly connected [Conn].
func Dial(rcvbuf int64) (*Conn, error) {
// kernel group is hard coded in lib/kobject_uevent.c, undocumented
c, err := netlink.Dial(syscall.NETLINK_KOBJECT_UEVENT, 1, rcvbuf)
if err != nil {
return nil, err
}
return &Conn{conn: c}, err
}
var (
// ErrBadSocket is returned by [Conn.Consume] for a reply from a
// syscall.Sockaddr with unexpected concrete type.
ErrBadSocket = errors.New("unexpected socket address")
)
// ReceiveBufferError indicates one or more [Message] being lost due to the
// socket receive buffer filling up. This is usually caused by epoll waking the
// receiving program up too late.
type ReceiveBufferError struct{ _ [0]*ReceiveBufferError }
var _ NeedsColdboot = ReceiveBufferError{}
func (ReceiveBufferError) recoverable() {}
func (ReceiveBufferError) coldboot() {}
func (ReceiveBufferError) Unwrap() error { return syscall.ENOBUFS }
func (e ReceiveBufferError) Error() string { return syscall.ENOBUFS.Error() }
// BadPortError is returned by [Conn.Consume] upon receiving a message that did
// not come from the kernel.
type BadPortError syscall.SockaddrNetlink
var _ Recoverable = new(BadPortError)
func (*BadPortError) recoverable() {}
func (e *BadPortError) Error() string {
return "unexpected message from port id " + strconv.Itoa(int(e.Pid)) +
" on NETLINK_KOBJECT_UEVENT"
}
// receiveEvent receives a single event and returns the address of its [Message].
func (c *Conn) receiveEvent(ctx context.Context) (*Message, error) {
data, _, from, err := c.conn.Recvmsg(ctx, 0)
if err != nil {
if errors.Is(err, syscall.ENOBUFS) {
return nil, ReceiveBufferError{}
}
return nil, err
}
// lib/kobject_uevent.c:
// set portid 0 to inform userspace message comes from kernel
if v, ok := from.(*syscall.SockaddrNetlink); !ok {
return nil, ErrBadSocket
} else if v.Pid != 0 {
return nil, (*BadPortError)(v)
}
var msg Message
if err = msg.UnmarshalBinary(data); err != nil {
return nil, err
}
return &msg, err
}
// Consume continuously receives and parses events from the kernel. It returns
// the first error it encounters.
//
// Callers must not restart event processing after a non-nil error that does not
// satisfy [Recoverable] is returned.
func (c *Conn) Consume(ctx context.Context, events chan<- *Message) error {
if err := c.enterExcl(exclConsume); err != nil {
return err
}
defer c.exitExcl(exclConsume)
for {
msg, err := c.receiveEvent(ctx)
if err != nil {
return err
}
events <- msg
}
}
|
